필사 모드: AI 智能体完全指南 — 用 LangChain、LangGraph、CrewAI 构建自主 AI 系统

输入用户目标
      ↓
  [制定计划]
  将目标拆解为子任务
      ↓
  [选择行动]
  决定下一步行动(使用哪个工具?)
      ↓
  [执行工具]
  执行所选工具
      ↓
  [观察结果]
  确认工具执行结果
      ↓
  [是否达成目标?] → 是 → 生成最终答案
       ↓ 否
  返回"选择行动"

Thought: 要回答这个问题，需要最新的股价数据
Action: 检索["三星电子 股价 2026"]
Observation: 三星电子现价 78,000 韩元，较前一交易日 +2.3%

Thought: 已获取股价数据，现在需要计算变动额
Action: 计算器[78000 * 0.023]
Observation: 1,794

Thought: 较前一日上涨了 1,794 韩元，现在可以作答了
Final Answer: 三星电子现价为 78,000 韩元，较前一日上涨 1,794 韩元(+2.3%)。

from langchain import hub
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.prompts import PromptTemplate

# ReAct 提示词模板
REACT_TEMPLATE = """你是一个乐于助人的 AI 智能体。
你可以使用以下工具:

{tools}

请按以下格式作答:

Question: 需要回答的输入问题
Thought: 始终思考接下来该做什么
Action: 要使用的工具。必须是 [{tool_names}] 中的一个
Action Input: 输入给工具的内容
Observation: 工具的结果
...(这个 Thought/Action/Observation 过程可重复 N 次)
Thought: 我现在知道最终答案了
Final Answer: 针对原始问题的最终答案

开始吧!

Question: {input}
Thought:{agent_scratchpad}"""

react_prompt = PromptTemplate.from_template(REACT_TEMPLATE)

# 配置 LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 定义工具
search = DuckDuckGoSearchRun()
tools = [
    Tool(
        name="Search",
        func=search.run,
        description="检索最新信息。适用于查询时事、最新数据"
    ),
    Tool(
        name="Calculator",
        func=lambda x: eval(x),  # 实际环境中应使用安全的计算器
        description="数学计算。输入为 Python 表达式"
    )
]

# 创建 ReAct 智能体
agent = create_react_agent(llm, tools, react_prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    handle_parsing_errors=True
)

# 执行
result = agent_executor.invoke({
    "input": "2026年3月现在比特币价格是多少，与去年3月相比涨跌幅是多少?"
})
print(result["output"])

from openai import OpenAI
import json

client = OpenAI()

# 函数定义
functions = [
    {
        "name": "get_weather",
        "description": "获取指定城市的当前天气",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "要查询天气的城市名称"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度单位"
                }
            },
            "required": ["city"]
        }
    },
    {
        "name": "search_database",
        "description": "在内部数据库中检索信息",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "检索查询"
                },
                "table": {
                    "type": "string",
                    "enum": ["users", "products", "orders"],
                    "description": "要检索的表"
                }
            },
            "required": ["query"]
        }
    }
]

# 实际函数实现
def get_weather(city: str, unit: str = "celsius") -> dict:
    """调用天气 API(实际实现)"""
    # 实际场景中会调用 Weather API
    return {
        "city": city,
        "temperature": 15,
        "unit": unit,
        "condition": "晴天",
        "humidity": 60
    }

def search_database(query: str, table: str = "products") -> list:
    """数据库检索(实际实现)"""
    # 实际场景中会执行数据库查询
    return [{"id": 1, "name": "示例产品", "price": 10000}]

# 工具映射
available_tools = {
    "get_weather": get_weather,
    "search_database": search_database
}

def run_agent_with_tools(user_message: str) -> str:
    """执行 Function Calling 智能体"""
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=[{"type": "function", "function": f} for f in functions],
            tool_choice="auto"
        )

        message = response.choices[0].message

        # 没有工具调用时返回最终答案
        if not message.tool_calls:
            return message.content

        # 处理工具调用
        messages.append(message)

        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            print(f"  工具调用: {func_name}({func_args})")

            # 执行函数
            if func_name in available_tools:
                result = available_tools[func_name](**func_args)
            else:
                result = {"error": f"未知函数: {func_name}"}

            # 把工具结果加入消息
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })

# 执行示例
answer = run_agent_with_tools("告诉我首尔现在的天气，并判断是否需要带伞")
print(answer)

from langchain.tools import tool
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
import subprocess
import sqlite3

@tool
def calculator(expression: str) -> str:
    """执行数学计算。请输入 Python 表达式。"""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

@tool
def run_python_code(code: str) -> str:
    """执行 Python 代码并返回结果。"""
    try:
        # 安全的执行环境(实际场景中建议使用 sandbox)
        local_vars = {}
        exec(code, {"__builtins__": {}}, local_vars)
        output = local_vars.get('result', 'No result variable found')
        return str(output)
    except Exception as e:
        return f"代码执行错误: {e}"

@tool
def query_database(sql: str) -> str:
    """在 SQLite 数据库中执行 SQL 查询。"""
    try:
        conn = sqlite3.connect("agent_db.sqlite")
        cursor = conn.cursor()
        cursor.execute(sql)
        rows = cursor.fetchall()
        conn.close()
        return str(rows)
    except Exception as e:
        return f"数据库错误: {e}"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件。"""
    # 实际场景中会使用 SMTP 或邮件 API
    print(f"发送邮件: {to}")
    print(f"主题: {subject}")
    print(f"内容: {body[:100]}...")
    return f"邮件已成功发送至 {to}。"

# Wikipedia 工具
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

@tool
def search_wikipedia(query: str) -> str:
    """在 Wikipedia 上检索信息。"""
    return wikipedia.run(query)

LangChain 架构
├── Models (LLM、Chat Models、Embeddings)
├── Prompts (PromptTemplate、ChatPromptTemplate)
├── Chains (LLMChain、SequentialChain、LCEL)
├── Memory (Buffer、Summary、VectorStore)
├── Agents (ReAct、OpenAI Functions)
├── Tools (内置 + 自定义)
└── Retrievers (VectorStore、MultiQuery)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnableParallel, RunnableLambda

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 基本链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位专业分析师。"),
    ("human", "{question}")
])

chain = prompt | llm | StrOutputParser()

result = chain.invoke({"question": "AI 智能体的优点是什么?"})
print(result)

# 结构化输出
from pydantic import BaseModel, Field

class AnalysisResult(BaseModel):
    summary: str = Field(description="摘要")
    key_points: list[str] = Field(description="核心要点列表")
    recommendation: str = Field(description="建议事项")

structured_chain = prompt | llm.with_structured_output(AnalysisResult)
result = structured_chain.invoke({"question": "比较 LangChain 与 LlamaIndex"})
print(result.summary)
print(result.key_points)

# 并行链
parallel_chain = RunnableParallel({
    "pros": ChatPromptTemplate.from_template("{topic}的优点是什么?") | llm | StrOutputParser(),
    "cons": ChatPromptTemplate.from_template("{topic}的缺点是什么?") | llm | StrOutputParser(),
})

result = parallel_chain.invoke({"topic": "AI 智能体"})
print("优点:", result["pros"])
print("缺点:", result["cons"])

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI

# 缓冲记忆(保留完整对话)
buffer_memory = ConversationBufferMemory(
    memory_key="history",
    return_messages=True
)

# 摘要记忆(把长对话进行摘要)
summary_memory = ConversationSummaryMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    memory_key="history",
    return_messages=True
)

# 对话链
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

conversation = ConversationChain(
    llm=llm,
    memory=buffer_memory,
    verbose=True
)

# 对话
r1 = conversation.predict(input="我的名字是金英柱")
r2 = conversation.predict(input="我刚才说我的名字是什么来着?")  # 会记住
print(r1, r2)

# 基于向量存储的长期记忆
from langchain.memory import VectorStoreRetrieverMemory
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
import faiss

# 初始化向量存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(["dummy"], embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

vector_memory = VectorStoreRetrieverMemory(retriever=retriever)
vector_memory.save_context(
    {"input": "我喜欢吃的食物是泡菜锅"},
    {"output": "明白了!"}
)

# 检索相关记忆
relevant = vector_memory.load_memory_variables({"prompt": "推荐点吃的"})
print(relevant)

from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain_core.prompts import ChatPromptTemplate

# 加载文档
loader = WebBaseLoader("https://example.com/document")
documents = loader.load()

# 文本分割
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = splitter.split_documents(documents)

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# RAG 链
llm = ChatOpenAI(model="gpt-4o", temperature=0)

rag_prompt = ChatPromptTemplate.from_template("""
请基于以下上下文回答问题。
上下文中没有的信息请回答不知道。

上下文:
{context}

问题: {question}

回答:""")

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    chain_type_kwargs={"prompt": rag_prompt},
    return_source_documents=True
)

result = qa_chain.invoke({"query": "主要内容是什么?"})
print("回答:", result["result"])
print("出处:", [doc.metadata for doc in result["source_documents"]])

from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferWindowMemory
from langchain.tools import tool
from langchain_community.tools import DuckDuckGoSearchRun
import datetime

# 工具定义
search = DuckDuckGoSearchRun()

@tool
def get_current_datetime() -> str:
    """返回当前日期和时间。"""
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def calculate(expression: str) -> str:
    """执行数学计算。例如: 2+2、10*5、sqrt(16)"""
    import math
    safe_dict = {k: getattr(math, k) for k in dir(math) if not k.startswith('_')}
    safe_dict['abs'] = abs
    try:
        return str(eval(expression, {"__builtins__": {}}, safe_dict))
    except Exception as e:
        return f"计算错误: {e}"

@tool
def web_search(query: str) -> str:
    """在网上检索最新信息。"""
    return search.run(query)

tools = [get_current_datetime, calculate, web_search]

# 智能体提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一位乐于助人的 AI 助手。
请准确、友善地回答用户的问题。
必要时使用工具来收集信息。
请始终使用中文回答。"""),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 记忆
memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    return_messages=True,
    k=10  # 保留最近 10 轮对话
)

# 创建智能体
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True,
    max_iterations=5,
    handle_parsing_errors=True
)

# 执行对话
def chat(message: str) -> str:
    result = agent_executor.invoke({"input": message})
    return result["output"]

# 测试
print(chat("今天是几号?"))
print(chat("告诉我三星电子最近的股价走势"))
print(chat("刚才提到的那家公司市值是多少?"))  # 利用记忆

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from typing import TypedDict, Annotated, Sequence
import operator

# 定义状态模式
class AgentState(TypedDict):
    messages: Annotated[Sequence, operator.add]  # 消息累积
    next: str  # 下一个节点

# 设置模型与工具
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [web_search, calculate, get_current_datetime]
llm_with_tools = llm.bind_tools(tools)

# 定义节点函数
def agent_node(state: AgentState) -> AgentState:
    """由 LLM 决定下一步行动的节点"""
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    """决定是继续执行还是结束(条件边)"""
    messages = state["messages"]
    last_message = messages[-1]

    # 有工具调用则继续，没有则结束
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return "tools"
    return END

# 构建图
tool_node = ToolNode(tools)

workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)

# 入口点
workflow.set_entry_point("agent")

# 添加边
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        END: END
    }
)
workflow.add_edge("tools", "agent")  # 工具执行后返回智能体

# 编译图
app = workflow.compile()

# 执行
result = app.invoke({
    "messages": [HumanMessage(content="告诉我首尔的天气")]
})

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

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# 检查点存储(保存对话会话)
memory = MemorySaver()

class ApprovalState(TypedDict):
    messages: Annotated[Sequence, operator.add]
    pending_action: str
    approved: bool

def agent_node(state: ApprovalState) -> ApprovalState:
    """智能体提出行动建议"""
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)

    # 重要行动需要等待批准
    if hasattr(response, 'tool_calls') and response.tool_calls:
        tool_name = response.tool_calls[0]['name']
        if tool_name in ["send_email", "delete_file", "make_payment"]:
            return {
                "messages": [response],
                "pending_action": tool_name,
                "approved": False
            }

    return {"messages": [response]}

def human_approval_node(state: ApprovalState) -> ApprovalState:
    """人工批准节点(中断)"""
    # 执行会在此节点中断，等待人工输入
    print(f"\n需要批准的操作: {state['pending_action']}")
    print("要继续请输入 'approve'，要取消请输入 'reject'")
    # 实际场景中会通过 Web UI 或 Slack 发送通知
    return state

def check_approval(state: ApprovalState) -> str:
    if state.get("approved"):
        return "execute"
    elif state.get("pending_action") and not state.get("approved"):
        return "human_approval"
    return END

workflow = StateGraph(ApprovalState)
workflow.add_node("agent", agent_node)
workflow.add_node("human_approval", human_approval_node)
workflow.add_node("tools", ToolNode(tools))

workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", check_approval, {
    "human_approval": "human_approval",
    "execute": "tools",
    END: END
})
workflow.add_edge("tools", "agent")

# 设置中断点
app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["human_approval"]  # 在此节点之前中断
)

# 执行(会中断)
thread_id = "session_001"
config = {"configurable": {"thread_id": thread_id}}

result = app.invoke(
    {"messages": [HumanMessage(content="给团队成员发送会议邀请邮件")]},
    config=config
)

# 人工批准后恢复
app.update_state(config, {"approved": True})
final_result = app.invoke(None, config=config)  # None = 从之前的状态恢复

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing import TypedDict, Annotated, List
import operator

class ResearchState(TypedDict):
    topic: str
    search_queries: List[str]
    search_results: List[str]
    draft: str
    final_report: str
    iteration: int

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def plan_queries(state: ResearchState) -> ResearchState:
    """规划研究查询"""
    topic = state["topic"]

    response = llm.invoke([
        HumanMessage(content=f"""主题: {topic}
请为完整调研该主题生成 5 个检索查询。
以 JSON 格式返回: {{"queries": ["查询1", "查询2", ...]}}""")
    ])

    import json
    queries = json.loads(response.content)["queries"]
    return {"search_queries": queries}

def execute_searches(state: ResearchState) -> ResearchState:
    """执行并行检索"""
    queries = state["search_queries"]
    results = []

    for query in queries:
        result = search.run(query)
        results.append(f"[{query}]\n{result}")

    return {"search_results": results}

def write_draft(state: ResearchState) -> ResearchState:
    """撰写初稿"""
    topic = state["topic"]
    results = "\n\n".join(state["search_results"])

    response = llm.invoke([
        HumanMessage(content=f"""主题: {topic}

收集到的信息:
{results}

请基于以上信息撰写一份详细的研究报告初稿。""")
    ])

    return {"draft": response.content, "iteration": state.get("iteration", 0) + 1}

def review_and_improve(state: ResearchState) -> ResearchState:
    """审阅并改进初稿"""
    draft = state["draft"]

    response = llm.invoke([
        HumanMessage(content=f"""请审阅并改进以下研究报告初稿:

{draft}

改进事项:
1. 核实准确性
2. 改善逻辑脉络
3. 补充重要信息
4. 强化结论

请撰写最终报告。""")
    ])

    return {"final_report": response.content}

def should_improve(state: ResearchState) -> str:
    """判断是否还需要进一步改进"""
    if state.get("iteration", 0) < 2:
        return "improve"
    return "finalize"

# 研究工作流图
research_graph = StateGraph(ResearchState)

research_graph.add_node("plan_queries", plan_queries)
research_graph.add_node("execute_searches", execute_searches)
research_graph.add_node("write_draft", write_draft)
research_graph.add_node("review_and_improve", review_and_improve)

research_graph.set_entry_point("plan_queries")
research_graph.add_edge("plan_queries", "execute_searches")
research_graph.add_edge("execute_searches", "write_draft")
research_graph.add_conditional_edges(
    "write_draft",
    should_improve,
    {
        "improve": "execute_searches",  # 需要补充检索
        "finalize": "review_and_improve"
    }
)
research_graph.add_edge("review_and_improve", END)

research_app = research_graph.compile()

# 执行
result = research_app.invoke({
    "topic": "2026年 AI 智能体技术趋势",
    "search_queries": [],
    "search_results": [],
    "draft": "",
    "final_report": "",
    "iteration": 0
})

print(result["final_report"])

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool, QueryEngineTool
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.llms.openai import OpenAI
from llama_index.core.settings import Settings

# 设置 LLM
Settings.llm = OpenAI(model="gpt-4o", temperature=0)

# 定义 FunctionTool
def multiply(a: float, b: float) -> float:
    """将两个数相乘。"""
    return a * b

def add(a: float, b: float) -> float:
    """将两个数相加。"""
    return a + b

multiply_tool = FunctionTool.from_defaults(fn=multiply)
add_tool = FunctionTool.from_defaults(fn=add)

# QueryEngineTool(文档检索工具)
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(similarity_top_k=3)

query_tool = QueryEngineTool.from_defaults(
    query_engine=query_engine,
    name="knowledge_base",
    description="在公司内部文档中检索信息。"
)

# ReAct 智能体
agent = ReActAgent.from_tools(
    [multiply_tool, add_tool, query_tool],
    llm=Settings.llm,
    verbose=True,
    max_iterations=10
)

# 执行
response = agent.chat("在内部文档中查找 AI 政策，并把违反政策的罚金 5 万韩元和 10 万韩元相加")
print(response)

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool
from llama_index.core import VectorStoreIndex, SummaryIndex
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core import SimpleDirectoryReader

# 加载多个文档集
docs_finance = SimpleDirectoryReader("./finance_docs").load_data()
docs_hr = SimpleDirectoryReader("./hr_docs").load_data()
docs_technical = SimpleDirectoryReader("./technical_docs").load_data()

# 为每个文档集创建索引
splitter = SentenceSplitter(chunk_size=512)

finance_index = VectorStoreIndex.from_documents(
    docs_finance, transformations=[splitter]
)
hr_index = VectorStoreIndex.from_documents(
    docs_hr, transformations=[splitter]
)
tech_index = VectorStoreIndex.from_documents(
    docs_technical, transformations=[splitter]
)

# 把每个索引转换为工具
tools = [
    QueryEngineTool.from_defaults(
        query_engine=finance_index.as_query_engine(),
        name="finance_qa",
        description="回答财务、会计、预算相关问题"
    ),
    QueryEngineTool.from_defaults(
        query_engine=hr_index.as_query_engine(),
        name="hr_qa",
        description="回答人事、招聘、福利相关问题"
    ),
    QueryEngineTool.from_defaults(
        query_engine=tech_index.as_query_engine(),
        name="tech_qa",
        description="回答技术规格、开发指南相关问题"
    ),
]

# 多文档智能体
agent = ReActAgent.from_tools(tools, verbose=True)

response = agent.chat("告诉我 2026 年的 IT 预算和新招聘计划")
print(response)

Crew (团队)
├── Agents (智能体们) - 各自拥有角色与目标
│   ├── Role (角色): "资深研究员"、"内容撰稿人"
│   ├── Goal (目标): 想要达成的事情
│   ├── Backstory (背景): 智能体的性格/专业性
│   └── Tools (工具): 可使用的工具
└── Tasks (任务们)
    ├── Description (描述): 需要完成的工作
    ├── Expected Output: 预期输出
    └── Agent: 负责的智能体

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# 工具
search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

# 定义智能体
researcher = Agent(
    role="资深研究员",
    goal="就给定主题收集全面而准确的信息",
    backstory="""你是一位拥有 10 年经验的专业研究员。擅长系统性地
    调研复杂主题并提炼核心洞见。
    始终从可信来源收集最新信息。""",
    tools=[search_tool, web_tool],
    llm=llm,
    verbose=True
)

analyst = Agent(
    role="数据分析师",
    goal="分析收集到的信息，找出模式与趋势",
    backstory="""你是一位数据分析专家。擅长从原始数据中
    发现有意义的模式，得出可执行的洞见。
    结合统计方法与商业知识进行分析。""",
    tools=[search_tool],
    llm=llm,
    verbose=True
)

writer = Agent(
    role="内容撰稿人",
    goal="把分析结果写成清晰且有说服力的报告",
    backstory="""你是一位专业撰稿人，擅长把技术性内容
    讲解得连普通读者也能理解，喜欢用讲故事的方式
    传达复杂的分析结果。""",
    llm=llm,
    verbose=True
)

# 定义任务
research_task = Task(
    description="""针对 '{topic}' 调研以下内容:
    1. 最新动态与发展
    2. 主要参与者及其打法
    3. 潜在机会与风险因素
    4. 相关统计数据

    请至少引用 5 个可信来源。""",
    expected_output="调研结果摘要(至少 500 字)",
    agent=researcher
)

analysis_task = Task(
    description="""分析研究员收集到的信息:
    1. 识别 3 个核心趋势
    2. 进行 SWOT 分析
    3. 给出 6 个月~1 年的短期展望
    4. 主要风险因素

    请提供基于数据的客观分析。""",
    expected_output="分析报告(至少 400 字)",
    agent=analyst,
    context=[research_task]  # 利用调研任务的结果
)

report_task = Task(
    description="""综合调研与分析结果，撰写一份专业报告:

    报告结构:
    1. 摘要(Executive Summary)
    2. 现状分析
    3. 核心洞见
    4. 建议事项
    5. 结论

    请专业且有说服力地撰写。""",
    expected_output="完整报告(至少 800 字)",
    agent=writer,
    context=[research_task, analysis_task]
)

# 组建团队(顺序执行)
research_crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.sequential,  # 顺序执行
    verbose=True
)

# 执行
result = research_crew.kickoff(inputs={"topic": "2026年 AI 智能体市场分析"})
print(result)

from crewai import Agent, Task, Crew, Process
from crewai_tools import CodeInterpreterTool

code_interpreter = CodeInterpreterTool()

# 软件开发团队
product_manager = Agent(
    role="产品经理",
    goal="明确定义需求并制定开发计划",
    backstory="拥有 10 年经验的 PM，擅长连接技术需求与业务目标。",
    llm=llm,
    verbose=True
)

senior_developer = Agent(
    role="高级开发工程师",
    goal="编写高质量、可扩展的代码",
    backstory="全栈开发者，精通 Python、FastAPI、React。",
    tools=[code_interpreter],
    llm=llm,
    verbose=True
)

qa_engineer = Agent(
    role="QA 工程师",
    goal="彻底测试代码并保证质量",
    backstory="软件测试专家，热衷于发现 Bug。",
    tools=[code_interpreter],
    llm=llm,
    verbose=True
)

# 开发任务
requirements_task = Task(
    description="""为 '{feature_request}' 定义技术需求:
    1. 用户故事
    2. 功能需求列表
    3. 非功能需求(性能、安全)
    4. API 设计(端点列表)""",
    expected_output="需求文档",
    agent=product_manager
)

development_task = Task(
    description="""基于需求文档编写 Python FastAPI 代码:
    1. 完整的 API 实现
    2. 数据模型(Pydantic)
    3. 错误处理
    4. 代码注释""",
    expected_output="完成的 Python 代码",
    agent=senior_developer,
    context=[requirements_task]
)

testing_task = Task(
    description="""审阅并测试编写好的代码:
    1. 代码审查(Bug、安全漏洞)
    2. 编写单元测试
    3. 测试边缘情况
    4. 提出改进建议""",
    expected_output="测试报告与改进后的代码",
    agent=qa_engineer,
    context=[development_task]
)

# 开发团队
dev_crew = Crew(
    agents=[product_manager, senior_developer, qa_engineer],
    tasks=[requirements_task, development_task, testing_task],
    process=Process.sequential,
    verbose=True
)

result = dev_crew.kickoff(
    inputs={"feature_request": "用户认证 API(基于 JWT)"}
)
print(result)

# 由管理者委派任务的层级结构
manager = Agent(
    role="项目经理",
    goal="协调团队并产出最佳成果",
    backstory="经验丰富的 PM，善于最大化发挥团队成员的优势。",
    llm=llm,
    verbose=True,
    allow_delegation=True  # 可委派给其他智能体
)

hierarchical_crew = Crew(
    agents=[manager, researcher, analyst, writer],
    tasks=[report_task],  # 只定义最终任务(其余自动分配)
    process=Process.hierarchical,  # 层级式执行
    manager_agent=manager,
    verbose=True
)

from langchain.memory import (
    ConversationBufferMemory,
    ConversationSummaryBufferMemory,
    ConversationEntityMemory,
)
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
embeddings = OpenAIEmbeddings()

# 1. 短期记忆 - 最近 N 条消息
from langchain.memory import ConversationBufferWindowMemory
short_term = ConversationBufferWindowMemory(k=5, return_messages=True)

# 2. 摘要记忆 - 把久远的对话进行摘要
summary_memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=1000,
    return_messages=True
)

# 3. 实体记忆 - 提取核心信息
entity_memory = ConversationEntityMemory(
    llm=llm,
    return_messages=True
)

# 4. 长期记忆 - 向量数据库
class LongTermMemory:
    def __init__(self):
        self.vectorstore = FAISS.from_texts(["init"], embeddings)
        self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": 5})

    def save(self, text: str, metadata: dict = None):
        """把重要信息保存到长期记忆"""
        self.vectorstore.add_texts([text], metadatas=[metadata or {}])

    def recall(self, query: str) -> list:
        """检索相关记忆"""
        docs = self.retriever.get_relevant_documents(query)
        return [doc.page_content for doc in docs]

# 5. 情景记忆 - 过往经验
class EpisodicMemory:
    """保存智能体过往的成功/失败经验"""
    def __init__(self):
        self.episodes = []

    def save_episode(self, task: str, actions: list, result: str, success: bool):
        episode = {
            "task": task,
            "actions": actions,
            "result": result,
            "success": success,
            "timestamp": datetime.datetime.now().isoformat()
        }
        self.episodes.append(episode)

    def find_similar_episodes(self, current_task: str) -> list:
        """检索相似的过往经验"""
        # 实际场景中会用向量相似度检索
        return [e for e in self.episodes if e["success"]]


# 综合记忆系统
class AgentMemorySystem:
    def __init__(self):
        self.short_term = ConversationBufferWindowMemory(k=10)
        self.long_term = LongTermMemory()
        self.episodic = EpisodicMemory()
        self.entities = {}

    def save_message(self, role: str, content: str):
        self.short_term.save_context(
            {"input": content if role == "human" else ""},
            {"output": content if role == "ai" else ""}
        )

    def extract_entities(self, text: str) -> dict:
        """从文本中提取核心信息(姓名、日期、数字等)"""
        # 实际场景中会用 NER 模型
        return {}

    def get_relevant_context(self, query: str) -> str:
        """检索相关记忆"""
        recent = self.short_term.load_memory_variables({})
        long_term = self.long_term.recall(query)
        past_episodes = self.episodic.find_similar_episodes(query)

        context = f"""最近对话: {recent.get('history', '')}

相关记忆: {'; '.join(long_term[:3])}

相似经验: {past_episodes[:2] if past_episodes else '无'}"""

        return context

memory_system = AgentMemorySystem()

from langchain_experimental.tools import PythonREPLTool
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

python_repl = PythonREPLTool()

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 数据分析智能体
data_analysis_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一位专业数据分析师。
熟练使用 Python、pandas、matplotlib、seaborn。
收到数据分析请求后，编写并执行代码来得出结果。
请始终连同代码一起说明分析结果。"""),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_openai_tools_agent(
    llm,
    [python_repl],
    data_analysis_prompt
)
data_agent = AgentExecutor(agent=agent, tools=[python_repl], verbose=True)

# 执行示例
result = data_agent.invoke({
    "input": """请分析以下数据:
    sales = [100, 150, 120, 200, 180, 250, 220, 300, 280, 350, 320, 400]
    months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun',
              'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']

    请分析月度销售趋势并计算同比增长率""",
    "chat_history": []
})
print(result["output"])

import docker
import tempfile
import os

class DockerCodeExecutor:
    """在 Docker 容器中安全执行代码"""

    def __init__(self, image="python:3.11-slim", timeout=30):
        self.client = docker.from_env()
        self.image = image
        self.timeout = timeout

    def execute(self, code: str, packages: list = None) -> dict:
        """
        在 Docker 容器中执行 Python 代码
        Returns: {success: bool, output: str, error: str}
        """
        with tempfile.TemporaryDirectory() as tmpdir:
            # 保存代码文件
            code_file = os.path.join(tmpdir, "script.py")
            with open(code_file, "w") as f:
                f.write(code)

            # 安装依赖包的命令
            install_cmd = ""
            if packages:
                pkgs = " ".join(packages)
                install_cmd = f"pip install {pkgs} -q && "

            try:
                container = self.client.containers.run(
                    self.image,
                    command=f'sh -c "{install_cmd}python /code/script.py"',
                    volumes={tmpdir: {"bind": "/code", "mode": "ro"}},
                    remove=True,
                    network_mode="none",  # 阻断网络
                    mem_limit="256m",      # 限制内存
                    cpu_period=100000,
                    cpu_quota=50000,       # CPU 限制为 50%
                    timeout=self.timeout,
                    stdout=True,
                    stderr=True
                )
                return {
                    "success": True,
                    "output": container.decode("utf-8"),
                    "error": ""
                }
            except docker.errors.ContainerError as e:
                return {
                    "success": False,
                    "output": "",
                    "error": e.stderr.decode("utf-8") if e.stderr else str(e)
                }
            except Exception as e:
                return {"success": False, "output": "", "error": str(e)}


executor = DockerCodeExecutor()

code = """
import pandas as pd
import json

data = {'姓名': ['Alice', 'Bob', 'Charlie'], '得分': [85, 92, 78]}
df = pd.DataFrame(data)
result = df.describe().to_dict()
print(json.dumps(result, ensure_ascii=False, indent=2))
"""

result = executor.execute(code, packages=["pandas"])
print(result["output"])

import os
from langsmith import Client
from langsmith.run_helpers import traceable

# 配置 LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "ai-agent-evaluation"

@traceable(run_type="chain")
def run_agent_with_tracking(user_input: str):
    """执行经由 LangSmith 追踪的智能体"""
    result = agent_executor.invoke({"input": user_input})
    return result

# 用 LangSmith 客户端查询执行数据
client = Client()

# 查询最近的执行数据
runs = client.list_runs(
    project_name="ai-agent-evaluation",
    run_type="chain"
)

for run in list(runs)[:5]:
    print(f"Run ID: {run.id}")
    print(f"状态: {run.status}")
    print(f"执行时间: {run.end_time - run.start_time if run.end_time else 'N/A'}")
    print(f"Token 使用量: {run.total_tokens}")
    print("---")

import time
import json
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class AgentMetrics:
    """智能体执行指标"""
    task: str
    success: bool
    total_time: float
    num_iterations: int
    tools_used: List[str]
    tokens_used: int
    error_message: Optional[str] = None
    final_answer: Optional[str] = None

class AgentEvaluator:
    """智能体性能评估系统"""

    def __init__(self, agent_executor):
        self.agent = agent_executor
        self.metrics_history: List[AgentMetrics] = []

    def evaluate(self, task: str, expected_keywords: list = None) -> AgentMetrics:
        """评估单个任务"""
        start_time = time.time()
        tools_used = []
        iterations = 0

        try:
            result = self.agent.invoke({"input": task})
            total_time = time.time() - start_time
            answer = result.get("output", "")

            # 判断是否成功
            success = True
            if expected_keywords:
                success = any(kw.lower() in answer.lower() for kw in expected_keywords)

            metrics = AgentMetrics(
                task=task,
                success=success,
                total_time=total_time,
                num_iterations=iterations,
                tools_used=tools_used,
                tokens_used=0,
                final_answer=answer
            )

        except Exception as e:
            metrics = AgentMetrics(
                task=task,
                success=False,
                total_time=time.time() - start_time,
                num_iterations=0,
                tools_used=[],
                tokens_used=0,
                error_message=str(e)
            )

        self.metrics_history.append(metrics)
        return metrics

    def batch_evaluate(self, test_cases: list) -> dict:
        """批量评估"""
        results = []
        for case in test_cases:
            task = case["task"]
            keywords = case.get("expected_keywords", [])
            metrics = self.evaluate(task, keywords)
            results.append(metrics)

        # 汇总统计
        successes = [r for r in results if r.success]
        success_rate = len(successes) / len(results) if results else 0
        avg_time = sum(r.total_time for r in results) / len(results)

        return {
            "total_tasks": len(results),
            "success_rate": success_rate,
            "avg_response_time": avg_time,
            "failed_tasks": [r.task for r in results if not r.success],
            "detailed_results": results
        }

    def generate_report(self) -> str:
        """生成性能报告"""
        if not self.metrics_history:
            return "无评估数据"

        total = len(self.metrics_history)
        successes = sum(1 for m in self.metrics_history if m.success)
        avg_time = sum(m.total_time for m in self.metrics_history) / total

        report = f"""
=== 智能体性能报告 ===
总任务数: {total}
成功率: {successes/total*100:.1f}%
平均响应时间: {avg_time:.2f}秒

失败任务:
"""
        for m in self.metrics_history:
            if not m.success:
                report += f"  - {m.task}: {m.error_message or '未达到质量标准'}\n"

        return report


# 测试用例
test_cases = [
    {
        "task": "告诉我现在的时间",
        "expected_keywords": ["2026", "时", "分"]
    },
    {
        "task": "100 的平方根是多少?",
        "expected_keywords": ["10"]
    },
    {
        "task": "说明 AI 智能体的主要组成部分",
        "expected_keywords": ["LLM", "工具", "记忆"]
    }
]

evaluator = AgentEvaluator(agent_executor)
report = evaluator.batch_evaluate(test_cases)
print(f"成功率: {report['success_rate']*100:.1f}%")
print(f"平均响应时间: {report['avg_response_time']:.2f}秒")