AI Agent开发实战【2026】- 从概念到落地

# ReAct模式示例
def react_loop(agent, goal):
    while not goal_achieved:
        # 思考：分析当前状态
        thought = agent.think(current_state)
        # 行动：选择并执行工具
        action = agent.select_action(thought)
        # 观察：获取执行结果
        observation = agent.execute(action)
        # 更新状态
        current_state = update_state(observation)

# 向量数据库实现长期记忆
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

class AgentMemory:
    def __init__(self):
        self.short_term = []  # 最近N轮对话
        self.long_term = Chroma(embedding_function=OpenAIEmbeddings())
    
    def store(self, content, memory_type="short"):
        if memory_type == "short":
            self.short_term.append(content)
            if len(self.short_term) > 10:
                self.short_term.pop(0)
        else:
            self.long_term.add_texts([content])
    
    def retrieve(self, query, k=5):
        return self.long_term.similarity_search(query, k=k)

from langchain.tools import Tool, tool

@tool
def search_web(query: str) -> str:
    """搜索网络获取最新信息"""
    # 实现搜索逻辑
    return search_results

@tool  
def execute_python(code: str) -> str:
    """执行Python代码并返回结果"""
    # 安全执行代码
    return exec_result

tools = [search_web, execute_python]

def reflect(agent, task, result):
    reflection_prompt = f"""
    任务：{task}
    执行结果：{result}
    
    请分析：
    1. 结果是否完全满足任务要求？
    2. 执行过程中有哪些可以优化的地方？
    3. 如果需要重新执行，应该如何调整策略？
    """
    return agent.llm.invoke(reflection_prompt)

from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain import hub

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

tools = [
    Tool(
        name="Search",
        func=lambda q: search_api(q),
        description="搜索网络获取信息"
    ),
    Tool(
        name="Calculator",
        func=lambda expr: eval(expr),
        description="执行数学计算"
    )
]

prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({
    "input": "查询今天上海的天气，并计算华氏温度"
})

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role='研究员',
    goal='收集和分析市场数据',
    backstory='你是一位经验丰富的市场研究专家',
    tools=[search_tool, scrape_tool],
    llm=llm
)

analyst = Agent(
    role='分析师',
    goal='基于研究数据生成洞察报告',
    backstory='你是一位数据分析专家，擅长发现趋势',
    tools=[analysis_tool],
    llm=llm
)

writer = Agent(
    role='撰稿人',
    goal='将分析结果转化为易读的报告',
    backstory='你是一位专业的商业写作专家',
    llm=llm
)

research_task = Task(
    description='研究2024年AI Agent市场趋势',
    agent=researcher,
    expected_output='市场数据和关键发现'
)

analysis_task = Task(
    description='分析市场数据，识别主要趋势',
    agent=analyst,
    expected_output='趋势分析报告'
)

report_task = Task(
    description='撰写最终市场分析报告',
    agent=writer,
    expected_output='完整的市场分析报告'
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.sequential
)

result = crew.kickoff()

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_step: str

def plan_step(state: AgentState):
    # 规划下一步行动
    return {"next_step": "execute", "messages": ["计划完成"]}

def execute_step(state: AgentState):
    # 执行行动
    return {"next_step": "reflect", "messages": ["执行完成"]}

def reflect_step(state: AgentState):
    # 反思结果
    if task_complete:
        return {"next_step": "end", "messages": ["任务完成"]}
    return {"next_step": "plan", "messages": ["需要重新规划"]}

workflow = StateGraph(AgentState)

workflow.add_node("plan", plan_step)
workflow.add_node("execute", execute_step)
workflow.add_node("reflect", reflect_step)

workflow.set_entry_point("plan")
workflow.add_edge("plan", "execute")
workflow.add_edge("execute", "reflect")
workflow.add_conditional_edges(
    "reflect",
    lambda x: x["next_step"],
    {"plan": "plan", "end": END}
)

app = workflow.compile()

system_prompt = """
你是一个专业的任务执行Agent。

## 工作原则
1. 在行动前先分析任务，制定清晰的计划
2. 每次只执行一个工具调用
3. 仔细观察工具返回结果
4. 如果结果不符合预期，反思原因并调整策略
5. 完成任务后，总结执行过程

## 可用工具
{tools_description}

## 输出格式
思考：[分析当前状态和下一步计划]
行动：[选择的工具和参数]
"""

class RobustAgent:
    def __init__(self, max_retries=3):
        self.max_retries = max_retries
    
    def execute_with_retry(self, task):
        for attempt in range(self.max_retries):
            try:
                result = self.execute(task)
                if self.validate_result(result):
                    return result
                # 结果不满足要求，反思并重试
                self.reflect_and_adjust(task, result)
            except Exception as e:
                self.handle_error(e, attempt)
        return self.fallback_response(task)