DeepSeek Agent Development Tutorial

pip install langchain langchain-openai langchain-community

from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain.prompts import PromptTemplate from langchain_community.tools import WikipediaQueryRun from langchain_community.utilities import WikipediaAPIWrapper import requests # Initialize DeepSeek model llm = ChatOpenAI( model="deepseek-chat", base_url="https://api.deepseek.com/v1", api_key="sk-your-api-key", temperature=0.1, ) # ========== Custom Tool Development ========== # Tool 1: Exchange Rate Query def get_exchange_rate(currency_pair: str) -> str: """Query real-time exchange rate. Input format: USD/CNY or EUR/USD""" try: base, target = currency_pair.split("/") url = f"https://api.exchangerate-api.com/v4/latest/{base}" data = requests.get(url, timeout=5).json() rate = data["rates"][target] return f"1 {base} = {rate} {target}" except Exception as e: return f"Exchange rate query failed: {e}" # Tool 2: File Reading def read_file(filepath: str) -> str: """Read local file content""" try: with open(filepath, "r", encoding="utf-8") as f: content = f.read() if len(content) > 2000: return content[:2000] + "\n... (content too long, truncated)" return content except Exception as e: return f"Failed to read file: {e}" # Tool 3: Python Code Execution def execute_python(code: str) -> str: """Safely execute Python code and return the result""" # Note: In production, use a sandbox environment (e.g., Docker) import io, sys old_stdout = sys.stdout sys.stdout = buffer = io.StringIO() try: exec(code, {"__builtins__": { "print": print, "range": range, "len": len, "int": int, "str": str, "list": list, "dict": dict, "sum": sum, "sorted": sorted, "enumerate": enumerate, }}) result = buffer.getvalue() return result.strip() if result.strip() else "Code executed successfully (no output)" except Exception as e: return f"Execution error: {e}" finally: sys.stdout = old_stdout # ========== Register Tools ========== tools = [ Tool( name="Exchange Rate Query", func=get_exchange_rate, description="Query real-time exchange rates. Input format: USD/CNY or EUR/USD", ), Tool( name="File Reading", func=read_file, description="Read local file content. Input is the file path", ), Tool( name="Python Execution", func=execute_python, description="Execute Python code and return the result. Input is a Python code string", ), Tool( name="Wikipedia", func=WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(lang="zh")).run, description="Search Chinese Wikipedia. Input is the search keyword", ), ] # ReAct Prompt Template prompt = PromptTemplate.from_template("""You are an intelligent assistant that can use the following tools to complete tasks: {tools} Use the following format to answer: Question: the user's question Thought: think about what to do next Action: tool name Action Input: tool input parameters Observation: tool return result ... (can repeat Thought/Action/Action Input/Observation) Thought: I now have enough information Final Answer: final answer Begin! Question: {input} Thought: {agent_scratchpad}""") # Create Agent and Executor agent = create_react_agent(llm=llm, tools=tools, prompt=prompt) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # Print detailed execution process handle_parsing_errors=True, # Automatically handle format errors max_iterations=8, # Maximum number of iterations ) # Run Agent result = agent_executor.invoke({ "input": "Query USD/CNY exchange rate, then calculate how many RMB 1000 USD can be exchanged for", }) print(f"\nFinal answer: {result['output']}")

Parameter	Description	Recommended Value
verbose	Whether to print detailed execution process	True during development, False in production
handle_parsing_errors	Automatically handle model output format errors	True
max_iterations	Maximum number of iterations to prevent infinite loops	5-10
early_stopping_method	Handling method after reaching maximum iterations	"generate" (generate final answer)

• Agent: Defines role, goal, backstory, and available tools
• Task: Defines the specific work to be done, including description, expected output, and assigned Agent
• Crew: Organizes multiple Agents to collaborate sequentially or hierarchically to complete a set of tasks
• Process: Controls the execution order of tasks, supporting sequential and hierarchical

# pip install crewai crewai-tools from crewai import Agent, Task, Crew, Process from crewai import LLM # Configure DeepSeek as the underlying model deepseek_llm = LLM( model="deepseek/deepseek-chat", base_url="https://api.deepseek.com/v1", api_key="sk-your-api-key", ) # ========== Define Agent Roles ========== # Researcher: responsible for information collection and analysis researcher = Agent( role="Senior Researcher", goal="Conduct in-depth research on the specified topic, collect comprehensive and accurate information and data", backstory="""You are an experienced research analyst who has worked at a top consulting firm for 10 years. You excel at quickly gathering information, identifying key trends, and supporting viewpoints with data. Your reports are always well-structured and well-argued.""", llm=deepseek_llm, verbose=True, allow_delegation=False, ) # Writer: responsible for content creation writer = Agent( role="Senior Content Writer", goal="Write engaging, professional, and accurate content based on research materials", backstory="""You are a senior technology writer who has contributed to several well-known tech media outlets. You excel at transforming complex technical concepts into accessible articles. Your articles are logically clear, data-accurate, and highly readable.""", llm=deepseek_llm, verbose=True, allow_delegation=False, ) # Reviewer: responsible for quality control reviewer = Agent( role="Content Reviewer", goal="Review the accuracy, readability, and professionalism of content to ensure output quality", backstory="""You are a senior editor who has worked at a top publishing house for 15 years. You have extremely high standards for text and can spot the slightest errors. Your review criteria include: factual accuracy, logical coherence, language expression, and formatting.""", llm=deepseek_llm, verbose=True, allow_delegation=False, ) # ========== Define Tasks ========== research_task = Task( description="""Research the latest developments in DeepSeek Agent development, including: 1. DeepSeek API's Function Calling capabilities 2. Integration methods of LangChain and CrewAI with DeepSeek 3. Best practices for Agents in production environments 4. Comparison of Agent capabilities with other LLMs (GPT-4, Claude) Please provide detailed data and specific technical details.""", expected_output="A structured research report containing key findings, technical comparisons, and best practice recommendations, no less than 500 words", agent=researcher, ) writing_task = Task( description="""Based on the research report, write an introductory article for developers on DeepSeek Agent development. Requirements: 1. Language should be easy to understand, suitable for developers with basic Python knowledge 2. Include specific code examples and architecture diagram descriptions 3. Highlight DeepSeek's unique advantages 4. Keep the length between 800-1000 words""", expected_output="A complete, ready-to-publish technical article", agent=writer, ) review_task = Task( description="""Review the quality of the article, checking: 1. Whether technical details are accurate 2. Whether code examples are runnable 3. Whether logic is clear and coherent 4. Whether language expression is professional If there are issues, mark the specific locations and provide suggestions for improvement.""", expected_output="Review report, including pass/fail determination and specific modification suggestions", agent=reviewer, ) # ========== Assemble Crew and Execute ========== crew = Crew( agents=[researcher, writer, reviewer], tasks=[research_task, writing_task, review_task], process=Process.sequential, # Sequential execution verbose=True, ) result = crew.kickoff() print("\n=== Final Output ===\n", result)

Best Practices for Multi-Agent Collaboration

• Each Agent should only be responsible for one clear role to avoid overlapping responsibilities
• Task descriptions should be specific, including a clear expected_output format
• Research agents can be configured with search tools, and execution agents can be configured with code tools
• Hierarchical process is suitable for complex projects, but requires an additional manager Agent
• Set allow_delegation=False to prevent Agents from shifting tasks to each other

1. Plan (Planning Phase): The LLM analyzes the task and generates a detailed execution plan (list of steps)
2. Execute (Execution Phase): The Agent executes step by step according to the plan, observing results at each step
3. Replan (Replanning): If a step fails or the result does not meet expectations, dynamically adjust the subsequent plan
4. Finalize (Completion): Aggregate the results of all steps and output the final answer

"""DeepSeek Plan-and-Execute Agent""" from openai import OpenAI import json client = OpenAI( api_key="sk-your-api-key", base_url="https://api.deepseek.com/v1", ) # ========== Planning Phase ========== PLANNER_PROMPT = """You are a task planning expert. Please decompose the user's task into clear execution steps. Requirements: 1. Each step should be an atomic operation, not further divisible 2. There should be clear dependencies between steps 3. Consider possible exceptions and alternative plans 4. Output in JSON format Output format: { "task": "original task description", "steps": [ {"id": 1, "description": "step description", "depends_on": [], "expected_output": "expected output"}, {"id": 2, "description": "step description", "depends_on": [1], "expected_output": "expected output"} ] }""" def create_plan(task: str) -> dict: """Generate execution plan""" response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": PLANNER_PROMPT}, {"role": "user", "content": f"Please create an execution plan for the following task:\n{task}"}, ], temperature=0.1, ) content = response.choices[0].message.content # Extract JSON if "```json" in content: content = content.split("```json")[1].split("```")[0] return json.loads(content) # ========== Execution Phase ========== EXECUTOR_PROMPT = """You are a task execution expert. Please execute the current step and return the result. The complete plan for the current task: {plan} Historical execution results: {history} Current step: Step {step_id}: {step_description} Please execute this step and return the result. If the step fails, explain the reason.""" def execute_step(plan: dict, step: dict, history: list) -> str: """Execute a single step""" response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": EXECUTOR_PROMPT.format( plan=json.dumps(plan, ensure_ascii=False, indent=2), history=json.dumps(history, ensure_ascii=False, indent=2) if history else "None", step_id=step["id"], step_description=step["description"], )}, ], temperature=0.1, ) return response.choices[0].message.content # ========== Main Loop ========== def plan_and_execute(task: str): """Plan-and-Execute main function""" print("=" * 60) print(f"Task: {task}") # Phase 1: Planning print("\n[Planning Phase] Analyzing task...") plan = create_plan(task) print(f"Generated {len(plan['steps'])} execution steps:") for s in plan["steps"]: print(f" Step {s['id']}: {s['description']}") # Phase 2: Execution print("\n[Execution Phase] Starting execution...") history = [] for step in plan["steps"]: print(f"\n--- Executing step {step['id']}: {step['description']} ---") result = execute_step(plan, step, history) print(f"Result: {result[:200]}...") history.append({ "step_id": step["id"], "description": step["description"], "result": result, "status": "success" if "失败" not in result else "failed", }) # Phase 3: Summary print("\n[Summary Phase] Generating final report...") return history # Test result = plan_and_execute("Analyze DeepSeek API pricing strategy and write a competitive comparison report") print(f"\nExecution completed, total {len(result)} steps completed")

Dynamic Plan Adjustment

When a step fails, you can have the Agent re-plan the remaining steps. After execute_step returns failure, call create_plan with the history of completed steps to let the model generate a new execution plan. This adaptive capability is a hallmark of advanced Agents.

Protection Layer	Purpose	Implementation
Input Validation	Filter malicious inputs to prevent injection attacks	Regex filtering + content moderation API
Tool Permission Control	Restrict the tools and parameter ranges an Agent can invoke	Whitelist + parameter validation + rate limiting
Output Filtering	Filter sensitive information to prevent data leakage	Regex redaction + sensitive word filtering
Audit Logging	Record all Agent operations for traceability	Structured logs + database storage

import re import logging from functools import wraps from datetime import datetime # ========== 1. Input Validation ========== class InputGuard: """Input safety check""" # Dangerous patterns DANGEROUS_PATTERNS = [ r'rm\s+-rf', # delete command r'DROP\s+TABLE', # SQL injection r'__import__', # Python dangerous import r'os\.system', # system command r'eval\s*\(', # eval execution r'exec\s*\(', # exec execution ] MAX_INPUT_LENGTH = 10000 @classmethod def validate(cls, user_input: str) -> tuple[bool, str]: """Validate input, returns (is_safe, reason)""" if len(user_input) > cls.MAX_INPUT_LENGTH: return False, "Input length exceeds limit" for pattern in cls.DANGEROUS_PATTERNS: if re.search(pattern, user_input, re.IGNORECASE): return False, f"Dangerous pattern detected: {pattern}" return True, "OK" # ========== 2. Tool Permission Control ========== class ToolGuard: """Tool call safety control""" # Tool whitelist ALLOWED_TOOLS = {"search", "calculate", "get_weather"} # Call count limit per tool tool_call_count = {} MAX_CALLS_PER_TOOL = 10 MAX_TOTAL_CALLS = 30 @classmethod def check_tool_access(cls, tool_name: str, args: dict) -> tuple[bool, str]: """Check if tool call is allowed""" if tool_name not in cls.ALLOWED_TOOLS: return False, f"Tool '{tool_name}' is not in the whitelist" cls.tool_call_count[tool_name] = cls.tool_call_count.get(tool_name, 0) + 1 if cls.tool_call_count[tool_name] > cls.MAX_CALLS_PER_TOOL: return False, f"Tool '{tool_name}' call count exceeded" total = sum(cls.tool_call_count.values()) if total > cls.MAX_TOTAL_CALLS: return False, "Total call count exceeded" return True, "OK" # ========== 3. Output Filtering ========== class OutputGuard: """Output security filtering""" # Sensitive information patterns SENSITIVE_PATTERNS = [ (re.compile(r'\b\d{15,19}\b'), '[Bank card number hidden]'), (re.compile(r'\b1[3-9]\d{9}\b'), '[Phone number hidden]'), (re.compile(r'\b[\w.-]+@[\w.-]+\.\w+\b'), '[Email hidden]'), (re.compile(r'\b\d{6}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]\b'), '[ID card number hidden]'), ] @classmethod def sanitize(cls, output: str) -> str: """Sanitize output content""" sanitized = output for pattern, replacement in cls.SENSITIVE_PATTERNS: sanitized = pattern.sub(replacement, sanitized) return sanitized # ========== 4. Audit Logging ========== logging.basicConfig( filename='agent_audit.log', level=logging.INFO, format='%(asctime)s | %(levelname)s | %(message)s', ) def log_agent_action(action: str, detail: dict): """Log Agent action""" logging.info(f"{action} | {json.dumps(detail, ensure_ascii=False)}") # ========== Safe Agent Wrapper ========== class SafeAgent: """Agent wrapper with safety guardrails""" def __init__(self, agent_executor): self.executor = agent_executor def run(self, user_input: str) -> str: # 1. Input validation safe, reason = InputGuard.validate(user_input) if not safe: log_agent_action("INPUT_REJECTED", {"reason": reason}) return f"Input rejected: {reason}" log_agent_action("INPUT_ACCEPTED", {"input": user_input[:100]}) # 2. Execute Agent try: result = self.executor.invoke({"input": user_input}) output = result["output"] except Exception as e: log_agent_action("EXECUTION_ERROR", {"error": str(e)}) return "Agent execution failed, please try again later." # 3. Output filtering safe_output = OutputGuard.sanitize(output) log_agent_action("OUTPUT_GENERATED", {"output": safe_output[:100]}) return safe_output

Security Best Practices

• Never give the Agent direct access to the file system or database; use restricted APIs instead
• Code execution tools must use a sandbox environment (Docker container or restricted Python)
• All external API calls must go through a proxy, limiting rate and access scope
• Regularly audit Agent logs to detect abnormal behavior promptly
• Use System Prompt to clearly inform the Agent of security boundaries and prohibited behaviors

# pip install fastapi uvicorn pydantic from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from typing import Optional import uvicorn import time app = FastAPI( title="DeepSeek Agent API", version="1.0.0", description="DeepSeek Agent intelligent agent service", ) # CORS configuration app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # ========== Data Models ========== class AgentRequest(BaseModel): query: str = Field(..., min_length=1, max_length=5000) session_id: Optional[str] = None max_steps: int = Field(default=5, ge=1, le=20) class AgentResponse(BaseModel): success: bool answer: str steps: int tools_used: list[str] execution_time: float session_id: Optional[str] # ========== API Endpoints ========== @app.get("/health") async def health_check(): return {"status": "healthy", "model": "deepseek-chat"} @app.post("/agent/run", response_model=AgentResponse) async def run_agent(request: AgentRequest): """Execute Agent task""" start_time = time.time() tools_used = [] try: # Call your Agent executor here # result = safe_agent.run(request.query) # Example return answer = f"Agent has processed the query: {request.query}" return AgentResponse( success=True, answer=answer, steps=3, tools_used=tools_used, execution_time=round(time.time() - start_time, 3), session_id=request.session_id, ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/agent/chat") async def chat(request: AgentRequest): """Streaming chat endpoint (SSE)""" from fastapi.responses import StreamingResponse import asyncio async def generate(): # Simulate streaming output chunks = ["Analyzing the problem...", "Calling tools...", "Generating answer..."] for chunk in chunks: yield f"data: {chunk}\n\n" await asyncio.sleep(0.5) yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream") # Start the service if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)

# Dockerfile FROM python:3.11-slim WORKDIR /app # Install dependencies COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # Copy application code COPY . . # Expose port EXPOSE 8000 # Start service CMD ["uvicorn", "agent_server:app", "--host", "0.0.0.0", "--port", "8000"] # requirements.txt # fastapi==0.115.0 # uvicorn[standard]==0.30.0 # openai==1.50.0 # langchain==0.3.0 # langchain-openai==0.2.0 # chromadb==0.5.0 # sentence-transformers==3.0.0 # pydantic==2.9.0 # docker-compose.yml # version: '3.8' # services: # agent: # build: . # ports: # - "8000:8000" # environment: # - DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY} # volumes: # - ./chroma_db:/app/chroma_db # - ./agent_audit.log:/app/agent_audit.log # restart: unless-stopped # Build and start # docker-compose up -d --build

# pip install prometheus-client from prometheus_client import Counter, Histogram, Gauge, generate_latest from fastapi import Response # Define monitoring metrics agent_requests = Counter( 'agent_requests_total', 'Total Agent requests', ['status'], ) agent_duration = Histogram( 'agent_request_duration_seconds', 'Agent request duration', buckets=[0.5, 1, 2, 5, 10, 30], ) tool_calls = Counter( 'agent_tool_calls_total', 'Tool call count', ['tool_name'], ) active_sessions = Gauge( 'agent_active_sessions', 'Active sessions', ) # Expose Prometheus metrics endpoint @app.get("/metrics") async def metrics(): return Response(content=generate_latest(), media_type="text/plain")

Category	Check Item	Tool/Solution
Security	Input validation, output filtering, tool permissions	SafeAgent wrapper
Monitoring	Request volume, latency, error rate, tool calls	Prometheus + Grafana
Logging	Request logs, tool call logs, error logs	ELK / Loki / Structured logs
Rate Limiting	API rate limiting, concurrency control	slowapi / Redis + token bucket
Fault Tolerance	Retry mechanism, degradation strategy, health checks	tenacity + /health endpoint
Deployment	Containerization, rolling updates, auto-scaling	Docker + K8s + HPA

Deployment Recommendations

For small-scale applications, a single-machine Docker Compose deployment is sufficient. For production-grade applications, it is recommended to use a K8s cluster deployment with Prometheus + Grafana monitoring and ELK log collection. The Agent service is stateless and can be scaled horizontally, but attention must be paid to the concurrency limits of the LLM API.