用 deepagents 构建长任务 Agent：规划、文件系统、子 Agent 与 Milvus 记忆

[
  {
    "task": "收集 Milvus 的核心架构资料",
    "status": "pending",
    "priority": "high"
  },
  {
    "task": "整理 Milvus 与 pgvector、Qdrant 的差异",
    "status": "pending",
    "priority": "high"
  },
  {
    "task": "生成适合生产环境选型的结论",
    "status": "pending",
    "priority": "medium"
  }
]

/tool_results/internet_search_1.txt

from deepagents import create_deep_agent
from tavily import TavilyClient

tavily_client = TavilyClient(api_key="YOUR_TAVILY_API_KEY")


def internet_search(query: str, max_results: int = 5) -> str:
    """Run a web search and return concise search results."""
    results = tavily_client.search(query=query, max_results=max_results)
    return "\n".join(
        f"{item['title']}: {item['content']}"
        for item in results["results"]
    )


agent = create_deep_agent(
    tools=[internet_search]
)

from langchain_mcp_adapters.client import MultiServerMCPClient
from deepagents import create_deep_agent


async def main():
    mcp_client = MultiServerMCPClient(
        {
            "filesystem": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
                "transport": "stdio",
            }
        }
    )

    mcp_tools = await mcp_client.get_tools()

    agent = create_deep_agent(
        tools=mcp_tools
    )

    async for chunk in agent.astream(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "检查 /tmp 目录下的项目结构并给出说明"
                }
            ]
        }
    ):
        chunk["messages"][-1].pretty_print()

from langchain_core.tools import tool
from deepagents import create_deep_agent
from deepagents.middleware import AgentMiddleware


@tool
def get_weather(city: str) -> str:
    """Get the weather in a city."""
    return f"The weather in {city} is sunny."


class WeatherMiddleware(AgentMiddleware):
    tools = [get_weather]


agent = create_deep_agent(
    middleware=[WeatherMiddleware()]
)

from deepagents import create_deep_agent


def internet_search(query: str) -> str:
    """Run a web search."""
    return "search results"


research_subagent = {
    "name": "research-agent",
    "description": "Used to research in-depth questions",
    "prompt": "You are an expert researcher. Use search tools and return concise findings.",
    "tools": [internet_search],
    "model": "openai:gpt-4o",  # 不填时默认使用主 Agent 模型
}

agent = create_deep_agent(
    subagents=[research_subagent]
)

from deepagents import CompiledSubAgent, create_deep_agent
from langgraph.prebuilt import create_react_agent


custom_graph = create_react_agent(
    model="openai:gpt-4o",
    tools=[internet_search],
    prompt="You are a data analysis specialist."
)

agent = create_deep_agent(
    subagents=[
        CompiledSubAgent(
            name="data-analyzer",
            description="Specialized agent for data analysis",
            runnable=custom_graph
        )
    ]
)

from langchain_core.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver


@tool
def delete_file(path: str) -> str:
    """Delete a file from the filesystem."""
    return f"Deleted {path}"


agent = create_deep_agent(
    tools=[delete_file],
    interrupt_on={
        "delete_file": {
            "allowed_decisions": ["approve", "edit", "reject"]
        }
    },
    checkpointer=MemorySaver()
)

from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend

agent = create_deep_agent(
    backend=FilesystemBackend(root_dir="/path/to/project")
)

pip install -U deepagents tavily-python langchain-milvus langchain-openai langgraph

export TAVILY_API_KEY="your-tavily-api-key"
export OPENAI_API_KEY="your-openai-api-key"
export MILVUS_URI="http://localhost:19530"

import os

from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from langchain_openai import OpenAIEmbeddings
from langchain_milvus.storage import MilvusStore


embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small"
)

milvus_store = MilvusStore(
    collection_name="agent_memories",
    embedding_service=embeddings,
    connection_args={
        "uri": os.environ["MILVUS_URI"]
    }
)

backend = CompositeBackend(
    default=StateBackend(),
    routes={
        "/memories/": StoreBackend(store=milvus_store),
        "/knowledge/": StoreBackend(store=milvus_store),
    }
)

from langgraph.store.memory import InMemoryStore
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend

backend = CompositeBackend(
    default=StateBackend(),
    routes={
        "/memories/": StoreBackend(store=InMemoryStore())
    }
)

import os
from tavily import TavilyClient

tavily_client = TavilyClient(
    api_key=os.environ["TAVILY_API_KEY"]
)


def internet_search(query: str, max_results: int = 5) -> str:
    """Search the internet and return compact results."""
    results = tavily_client.search(
        query=query,
        max_results=max_results
    )

    return "\n\n".join(
        f"Title: {item['title']}\nContent: {item['content']}"
        for item in results["results"]
    )

from deepagents import create_deep_agent

agent = create_deep_agent(
    tools=[internet_search],
    system_prompt=(
        "你是技术研究专家。执行复杂研究任务时，需要先使用待办事项规划步骤。"
        "搜索得到的大块材料应保存到文件中，再用 grep 和 read_file 读取关键片段。"
        "具有跨任务复用价值的结论、用户偏好、技术对比和决策依据，"
        "必须写入 /memories/ 目录，方便后续会话复用。"
    ),
    backend=backend
)

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "研究 Milvus 向量数据库的技术特点，并整理适合生产选型的结论"
            }
        ]
    }
)

print(result["messages"][-1].content)

具有跨任务复用价值的结论写入 /memories/
临时草稿写入 /workspace/
搜索原始结果写入 /tool_results/

/workspace/    当前任务工作区
/temp/         临时文件
/tool_results/ 工具调用大结果
/memories/     长期记忆
/knowledge/    领域知识

["approve", "edit", "reject"]