DeepSeek Agent 開発チュートリアル

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 # DeepSeek モデルの初期化 llm = ChatOpenAI( model="deepseek-chat", base_url="https://api.deepseek.com/v1", api_key="sk-your-api-key", temperature=0.1, ) # ========== カスタムツール開発 ========== # ツール1: 為替レート照会 def get_exchange_rate(currency_pair: str) -> str: """リアルタイム為替レートを照会。入力形式: USD/CNY または 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"為替レート照会失敗: {e}" # ツール 2: ファイル読み取り def read_file(filepath: str) -> str: """ローカルファイルの内容を読み取る""" try: with open(filepath, "r", encoding="utf-8") as f: content = f.read() if len(content) > 2000: return content[:2000] + "\n... (内容が長すぎるため、切り詰めました)" return content except Exception as e: return f"ファイルの読み取りに失敗: {e}" # ツール 3: Pythonコード実行 def execute_python(code: str) -> str: """Pythonコードを安全に実行し、結果を返す""" # 注意: 本番環境ではサンドボックス環境（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 "コードの実行に成功しました（出力なし）" except Exception as e: return f"実行エラー: {e}" finally: sys.stdout = old_stdout # ========== ツールの登録 ========== tools = [ Tool( name="為替レート照会", func=get_exchange_rate, description="リアルタイム為替レートを照会します。入力形式: USD/CNY または EUR/USD", ), Tool( name="ファイル読み取り", func=read_file, description="ローカルファイルの内容を読み取ります。入力はファイルパスです", ), Tool( name="Python実行", func=execute_python, description="Pythonコードを実行し、結果を返します。入力はPythonコードの文字列です", ), Tool( name="ウィキペディア", func=WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(lang="zh")).run, description="中国語版ウィキペディアを検索します。入力は検索キーワードです", ), ] # ReActプロンプトテンプレート prompt = PromptTemplate.from_template("""あなたは以下のツールを使ってタスクを完了できるインテリジェントアシスタントです: {tools} 以下の形式で回答してください: Question: ユーザーの質問 Thought: 次に何をすべきか考える Action: ツール名 Action Input: ツールの入力パラメータ Observation: ツールの戻り結果 ... (Thought/Action/Action Input/Observation を繰り返すことができます) Thought: 十分な情報が得られました Final Answer: 最終回答 開始! Question: {input} Thought: {agent_scratchpad}""") # Agentと実行器の作成 agent = create_react_agent(llm=llm, tools=tools, prompt=prompt) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # 詳細な実行プロセスを出力 handle_parsing_errors=True, # 形式エラーを自動処理 max_iterations=8, # 最大反復回数 ) # Agentの実行 result = agent_executor.invoke({ "input": "USD/CNYの為替レートを照会し、1000ドルが何人民元に換算できるか計算してください", }) print(f"\n最終回答: {result['output']}")

パラメータ	説明	推奨値
verbose	詳細な実行プロセスを出力するかどうか	開発時はTrue、本番ではFalse
handle_parsing_errors	モデル出力の形式エラーを自動処理する	True
max_iterations	無限ループを防ぐための最大反復回数	5-10
early_stopping_method	最大反復回数に達した後の処理方法	"generate"（最終回答を生成）

• エージェント：役割、目標、バックストーリー、利用可能なツールを定義します
• タスク：実行する具体的な作業を定義します。説明、期待される出力、割り当てられたエージェントを含みます
• クルー：複数のエージェントを組織し、順次または階層的に協調させて一連のタスクを完了します
• プロセス：タスクの実行順序を制御し、sequential（順次）とhierarchical（階層）をサポートします

# pip install crewai crewai-tools from crewai import Agent, Task, Crew, Process from crewai import LLM # DeepSeekを基盤モデルとして設定 deepseek_llm = LLM( model="deepseek/deepseek-chat", base_url="https://api.deepseek.com/v1", api_key="sk-your-api-key", ) # ========== エージェントの役割を定義 ========== # リサーチャー: 情報収集と分析を担当 researcher = Agent( role="シニアリサーチャー", goal="指定されたテーマについて深く調査し、包括的で正確な情報とデータを収集する", backstory="""あなたは10年間トップコンサルティング会社で働いてきた経験豊富なリサーチアナリストです。 情報を素早く収集し、重要なトレンドを特定し、データで見解を裏付けることに優れています。 あなたのレポートは常に構造が明確で、論拠が十分です。""", llm=deepseek_llm, verbose=True, allow_delegation=False, ) # ライター: コンテンツ作成を担当 writer = Agent( role="シニアコンテンツライター", goal="調査資料に基づいて、魅力的で専門的かつ正確なコンテンツを作成する", backstory="""あなたは複数の有名なテクノロジーメディアに寄稿してきたシニアテクノロジーライターです。 複雑な技術概念をわかりやすい記事に変換することに優れています。 あなたの記事は論理が明確で、データが正確で、読みやすいです。""", llm=deepseek_llm, verbose=True, allow_delegation=False, ) # レビュアー: 品質管理を担当 reviewer = Agent( role="コンテンツレビュアー", goal="コンテンツの正確性、読みやすさ、専門性をレビューし、出力品質を保証する", backstory="""あなたはトップ出版社で15年間働いてきたシニアエディターです。 文章に対して非常に高い基準を持ち、最も細かい誤りも見つけることができます。 あなたのレビュー基準には、事実の正確性、論理の一貫性、言語表現、フォーマットが含まれます。""", llm=deepseek_llm, verbose=True, allow_delegation=False, ) # ========== タスクを定義 ========== research_task = Task( description="""DeepSeekエージェント開発の最新動向を調査してください。以下を含みます： 1. DeepSeek APIのFunction Calling機能 2. LangChainとCrewAIのDeepSeekとの統合方法 3. 本番環境におけるエージェントのベストプラクティス 4. 他のLLM（GPT-4、Claude）とのエージェント機能の比較 詳細なデータと具体的な技術的詳細を提供してください。""", expected_output="主要な発見、技術比較、ベストプラクティスの推奨事項を含む構造化された調査レポート。500語以上", agent=researcher, ) writing_task = Task( description="""研究レポートに基づいて、開発者向けのDeepSeek Agent開発入門記事を執筆してください。 要件： 1. わかりやすい言葉で、Pythonの基礎知識がある開発者に適していること 2. 具体的なコード例とアーキテクチャ図の説明を含むこと 3. DeepSeekの独自の強みを強調すること 4. 文字数は800〜1000字に収めること""", expected_output="完全で、すぐに公開できる技術記事", agent=writer, ) review_task = Task( description="""記事の品質をレビューし、以下をチェックします： 1. 技術的な詳細が正確かどうか 2. コード例が実行可能かどうか 3. 論理が明確で一貫しているかどうか 4. 言語表現が専門的かどうか 問題がある場合は、具体的な場所を指摘し、修正提案を行ってください。""", expected_output="レビューレポート。合格/不合格の判定と具体的な修正提案を含む", agent=reviewer, ) # ========== Crewを編成して実行 ========== crew = Crew( agents=[researcher, writer, reviewer], tasks=[research_task, writing_task, review_task], process=Process.sequential, # 順次実行 verbose=True, ) result = crew.kickoff() print("\n=== 最終出力 ===\n", result)

マルチエージェント連携のベストプラクティス

• 各エージェントは明確な役割を1つだけ担当し、責任の重複を避ける
• タスクの説明は具体的にし、明確なexpected_output形式を含める
• リサーチエージェントには検索ツール、実行エージェントにはコードツールを設定できる
• 階層プロセスは複雑なプロジェクトに適しているが、追加のマネージャーエージェントが必要
• allow_delegation=Falseを設定して、エージェント間でタスクを押し付け合わないようにする

1. Plan（計画フェーズ）：LLMがタスクを分析し、詳細な実行計画（ステップリスト）を生成します
2. Execute（実行フェーズ）：Agentは計画に従って段階的に実行し、各ステップの結果を観察します
3. Replan（再計画）：ステップが失敗した場合や結果が期待に沿わない場合、後続の計画を動的に調整します
4. Finalize（完了）：すべてのステップの結果を集約し、最終回答を出力します

"""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", ) # ========== 計画フェーズ ========== PLANNER_PROMPT = """あなたはタスク計画の専門家です。ユーザーのタスクを明確な実行ステップに分解してください。 要件： 1. 各ステップは原子的な操作であり、それ以上分割できないこと 2. ステップ間には明確な依存関係があること 3. 異常な状況や代替案を考慮すること 4. JSON形式で出力すること 出力形式： { "task": "元のタスク説明", "steps": [ {"id": 1, "description": "ステップの説明", "depends_on": [], "expected_output": "期待される出力"}, {"id": 2, "description": "ステップの説明", "depends_on": [1], "expected_output": "期待される出力"} ] }""" def create_plan(task: str) -> dict: """実行計画を生成""" response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": PLANNER_PROMPT}, {"role": "user", "content": f"次のタスクの実行計画を作成してください：\n{task}"}, ], temperature=0.1, ) content = response.choices[0].message.content # JSONを抽出 if "```json" in content: content = content.split("```json")[1].split("```")[0] return json.loads(content) # ========== 実行フェーズ ========== EXECUTOR_PROMPT = """あなたはタスク実行の専門家です。現在のステップを実行し、結果を返してください。 現在のタスクの完全な計画： {plan} 過去の実行結果： {history} 現在のステップ： ステップ {step_id}: {step_description} このステップを実行して結果を返してください。ステップが失敗した場合は、理由を説明してください。""" def execute_step(plan: dict, step: dict, history: list) -> str: """単一のステップを実行""" 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 "なし", step_id=step["id"], step_description=step["description"], )}, ], temperature=0.1, ) return response.choices[0].message.content # ========== メインループ ========== def plan_and_execute(task: str): """Plan-and-Execute メイン関数""" print("=" * 60) print(f"タスク: {task}") # フェーズ1: 計画 print("\n[計画フェーズ] タスクを分析中...") plan = create_plan(task) print(f"{len(plan['steps'])} 個の実行ステップを生成:") for s in plan["steps"]: print(f" ステップ {s['id']}: {s['description']}") # フェーズ2: 実行 print("\n[実行フェーズ] 実行を開始...") history = [] for step in plan["steps"]: print(f"\n--- 実行ステップ {step['id']}: {step['description']} ---") result = execute_step(plan, step, history) print(f"結果: {result[:200]}...") history.append({ "step_id": step["id"], "description": step["description"], "result": result, "status": "success" if "失敗" not in result else "failed", }) # Phase 3: 集計 print("\n[集計フェーズ] 最終レポートを生成中...") return history # テスト result = plan_and_execute("DeepSeek APIの価格戦略を分析し、競合比較レポートを作成する") print(f"\n実行完了、合計 {len(result)} ステップ完了")

動的プラン調整

あるステップの実行が失敗した場合、Agentに残りのステップを再計画させることができます。execute_stepが失敗を返した後、完了したステップの履歴を渡してcreate_planを呼び出し、モデルに新しい実行計画を生成させます。この適応能力は高度なAgentの特徴です。

防御層	役割	実装方法
入力検証	悪意のある入力をフィルタリングし、インジェクション攻撃を防ぐ	正規表現フィルタ + コンテンツモデレーションAPI
ツール権限制御	エージェントが呼び出せるツールとパラメータ範囲を制限する	ホワイトリスト + パラメータ検証 + レート制限
出力フィルタリング	機密情報をフィルタリングし、データ漏洩を防ぐ	正規表現によるマスキング + 不適切な言葉のフィルタリング
監査ログ	すべてのエージェント操作を記録し、追跡可能性を確保する	構造化ログ + データベース保存

import re import logging from functools import wraps from datetime import datetime # ========== 1. 入力検証 ========== class InputGuard: """入力安全チェック""" # 危険なパターン DANGEROUS_PATTERNS = [ r'rm\s+-rf', # 削除コマンド r'DROP\s+TABLE', # SQLインジェクション r'__import__', # Pythonの危険なインポート r'os\.system', # システムコマンド r'eval\s*\(', # eval実行 r'exec\s*\(', # exec実行 ] MAX_INPUT_LENGTH = 10000 @classmethod def validate(cls, user_input: str) -> tuple[bool, str]: """入力を検証し、(安全かどうか, 理由) を返す""" if len(user_input) > cls.MAX_INPUT_LENGTH: return False, "入力長が制限を超えています" for pattern in cls.DANGEROUS_PATTERNS: if re.search(pattern, user_input, re.IGNORECASE): return False, f"危険なパターンを検出: {pattern}" return True, "OK" # ========== 2. ツール権限制御 ========== class ToolGuard: """ツール呼び出しの安全制御""" # ツールのホワイトリスト ALLOWED_TOOLS = {"search", "calculate", "get_weather"} # ツールごとの呼び出し回数制限 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]: """ツール呼び出しが許可されているか確認""" if tool_name not in cls.ALLOWED_TOOLS: return False, f"ツール '{tool_name}' はホワイトリストにありません" 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_name}' の呼び出し回数が上限を超えました" total = sum(cls.tool_call_count.values()) if total > cls.MAX_TOTAL_CALLS: return False, "総呼び出し回数が上限を超えました" return True, "OK" # ========== 3. 出力フィルタリング ========== class OutputGuard: """出力の安全フィルタリング""" # 機密情報パターン SENSITIVE_PATTERNS = [ (re.compile(r'\b\d{15,19}\b'), '[銀行カード番号は非表示]'), (re.compile(r'\b1[3-9]\d{9}\b'), '[電話番号は非表示]'), (re.compile(r'\b[\w.-]+@[\w.-]+\.\w+\b'), '[メールアドレスは非表示]'), (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'), '[身分証番号は非表示]'), ] @classmethod def sanitize(cls, output: str) -> str: """出力内容をサニタイズ""" sanitized = output for pattern, replacement in cls.SENSITIVE_PATTERNS: sanitized = pattern.sub(replacement, sanitized) return sanitized # ========== 4. 監査ログ ========== logging.basicConfig( filename='agent_audit.log', level=logging.INFO, format='%(asctime)s | %(levelname)s | %(message)s', ) def log_agent_action(action: str, detail: dict): """Agent操作ログを記録""" logging.info(f"{action} | {json.dumps(detail, ensure_ascii=False)}") # ========== 安全なAgentラッパー ========== class SafeAgent: """安全ガードレール付きAgentラッパー""" def __init__(self, agent_executor): self.executor = agent_executor def run(self, user_input: str) -> str: # 1. 入力検証 safe, reason = InputGuard.validate(user_input) if not safe: log_agent_action("INPUT_REJECTED", {"reason": reason}) return f"入力が拒否されました: {reason}" log_agent_action("INPUT_ACCEPTED", {"input": user_input[:100]}) # 2. 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の実行中にエラーが発生しました。後でもう一度お試しください。" # 3. 出力フィルタリング safe_output = OutputGuard.sanitize(output) log_agent_action("OUTPUT_GENERATED", {"output": safe_output[:100]}) return safe_output

セキュリティのベストプラクティス

• Agentにファイルシステムやデータベースへの直接アクセス権限を与えず、制限付きAPIを使用する
• コード実行ツールはサンドボックス環境（Dockerコンテナまたは制限付きPython）を使用する
• すべての外部API呼び出しはプロキシを経由し、レートとアクセス範囲を制限する
• Agentログを定期的に監査し、異常な動作を早期に発見する
• System Promptを使用して、Agentにセキュリティ境界と禁止行為を明確に伝える

# 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 智能体服务", ) # CORS設定 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # ========== データモデル ========== 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エンドポイント ========== @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): """Agentタスクを実行""" start_time = time.time() tools_used = [] try: # ここでAgent実行器を呼び出します # result = safe_agent.run(request.query) # サンプル返答 answer = f"Agentはクエリを処理しました: {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): """ストリーミングチャットエンドポイント（SSE）""" from fastapi.responses import StreamingResponse import asyncio async def generate(): # ストリーミング出力をシミュレート chunks = ["問題を分析中...", "ツールを呼び出し中...", "回答を生成中..."] 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") # サービス起動 if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)

# Dockerfile FROM python:3.11-slim WORKDIR /app # 依存関係をインストール COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # アプリケーションコードをコピー COPY . . # ポートを公開 EXPOSE 8000 # サービス起動 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 # ビルドと起動 # docker-compose up -d --build

# pip install prometheus-client from prometheus_client import Counter, Histogram, Gauge, generate_latest from fastapi import Response # 監視メトリクスの定義 agent_requests = Counter( 'agent_requests_total', 'Agent リクエスト総数', ['status'], ) agent_duration = Histogram( 'agent_request_duration_seconds', 'Agent リクエスト所要時間', buckets=[0.5, 1, 2, 5, 10, 30], ) tool_calls = Counter( 'agent_tool_calls_total', 'ツール呼び出し回数', ['tool_name'], ) active_sessions = Gauge( 'agent_active_sessions', 'アクティブセッション数', ) # Prometheus メトリクスエンドポイントの公開 @app.get("/metrics") async def metrics(): return Response(content=generate_latest(), media_type="text/plain")

カテゴリ	チェック項目	ツール/ソリューション
セキュリティ	入力検証、出力フィルタリング、ツール権限	SafeAgent ラッパー
監視	リクエスト量、レイテンシ、エラー率、ツール呼び出し	Prometheus + Grafana
ログ	リクエストログ、ツール呼び出しログ、エラーログ	ELK / Loki / 構造化ログ
レート制限	API レート制限、並行制御	slowapi / Redis + トークンバケット
フォールトトレランス	リトライ機構、デグラデーション戦略、ヘルスチェック	tenacity + /health エンドポイント
デプロイ	コンテナ化、ローリングアップデート、自動スケーリング	Docker + K8s + HPA

デプロイ推奨事項

小規模アプリケーションの場合、単一マシンでの Docker Compose デプロイで十分です。本番グレードのアプリケーションでは、Prometheus + Grafana による監視と ELK によるログ収集を備えた K8s クラスタデプロイを推奨します。Agent サービスはステートレスであり、水平展開が可能ですが、LLM API の並行制限に注意する必要があります。