【AI学习路线 11】AI Agent智能体 - 自主决策与工具调用

from openai import OpenAI
import json

client = OpenAI(api_key="your-api-key")

# 定义工具函数
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "搜索互联网获取信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词"
                    }
                },
                "required": ["query"]
            }
        }
    }
]

# 实现工具函数
def get_weather(city):
    # 模拟天气API
    weather_data = {
        "北京": {"temp": 25, "weather": "晴"},
        "上海": {"temp": 28, "weather": "多云"},
        "深圳": {"temp": 30, "weather": "晴"}
    }
    return json.dumps(weather_data.get(city, {"temp": 20, "weather": "未知"}))

def search_web(query):
    # 模拟搜索
    return f"关于'{query}'的搜索结果..."

# Agent对话
def agent_chat(user_message):
    messages = [{"role": "user", "content": user_message}]
    
    # 第一次调用
    response = client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )
    
    # 检查是否需要调用工具
    response_message = response.choices[0].message
    
    if response_message.tool_calls:
        # 执行工具调用
        for tool_call in response_message.tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)
            
            # 调用对应函数
            if function_name == "get_weather":
                result = get_weather(function_args["city"])
            elif function_name == "search_web":
                result = search_web(function_args["query"])
            
            # 添加工具结果到消息
            messages.append(response_message)
            messages.append({
                "tool_call_id": tool_call.id,
                "role": "tool",
                "name": function_name,
                "content": result
            })
        
        # 再次调用获取最终回答
        final_response = client.chat.completions.create(
            model="gpt-4",
            messages=messages
        )
        return final_response.choices[0].message.content
    
    return response_message.content

# 测试
answer = agent_chat("北京今天天气怎么样？")
print(answer)

from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

# 定义工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    # 实际应用中调用天气API
    return f"{city}今天天气晴朗，温度25度"

@tool  
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

@tool
def search_database(query: str) -> str:
    """搜索数据库"""
    # 模拟数据库查询
    return f"查询'{query}'的结果..."

# 创建Agent
llm = ChatOpenAI(model="gpt-4")
tools = [get_weather, calculate, search_database]

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手，可以使用工具来帮助用户。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

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

# 执行
result = agent_executor.invoke({"input": "北京天气怎么样？温度加上10是多少？"})
print(result["output"])

from langchain.agents import create_react_agent
from langchain import hub

# ReAct Prompt
react_prompt = """
你需要回答以下问题。可以使用以下工具：

{tools}

使用以下格式：

Question: 输入问题
Thought: 思考该做什么
Action: 工具名称，必须是 [{tool_names}] 之一
Action Input: 工具输入
Observation: 工具输出
... (重复 Thought/Action/Action Input/Observation)
Thought: 我知道最终答案了
Final Answer: 最终答案

开始！

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

# 创建ReAct Agent
prompt = ChatPromptTemplate.from_template(react_prompt)
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

from langchain_experimental.plan_and_execute import (
    PlanAndExecute,
    load_agent_executor,
    load_chat_planner
)

# 创建规划器
planner = load_chat_planner(llm)

# 创建执行器
executor = load_agent_executor(llm, tools, verbose=True)

# 创建Plan-and-Execute Agent
agent = PlanAndExecute(planner=planner, executor=executor, verbose=True)

# 执行复杂任务
result = agent.invoke("帮我规划一次日本旅行，包括东京和大阪，预算1万元")

class CustomAgent:
    """自定义Agent实现"""
    
    def __init__(self, llm, tools, max_iterations=10):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        self.max_iterations = max_iterations
    
    def run(self, task):
        """运行Agent"""
        memory = []
        
        for i in range(self.max_iterations):
            # 生成思考
            thought = self._think(task, memory)
            memory.append(f"Thought: {thought}")
            
            # 判断是否完成
            if "FINISH" in thought or "任务完成" in thought:
                return self._extract_answer(memory)
            
            # 选择行动
            action, action_input = self._decide_action(thought)
            
            if action == "finish":
                return action_input
            
            # 执行工具
            if action in self.tools:
                observation = self.tools[action].invoke(action_input)
                memory.append(f"Action: {action}({action_input})")
                memory.append(f"Observation: {observation}")
        
        return "达到最大迭代次数"
    
    def _think(self, task, memory):
        """思考下一步"""
        prompt = f"""
        任务: {task}
        
        已执行步骤:
        {chr(10).join(memory)}
        
        请思考下一步该做什么。如果任务已完成，回复"任务完成"。
        """
        return self.llm.invoke(prompt).content
    
    def _decide_action(self, thought):
        """决定执行哪个工具"""
        # 使用LLM选择工具和输入
        prompt = f"""
        基于思考: {thought}
        
        可用工具: {list(self.tools.keys())}
        
        请输出要执行的工具名称和输入，格式：
        工具名|输入
        """
        response = self.llm.invoke(prompt).content
        parts = response.split("|")
        return parts[0].strip(), parts[1].strip() if len(parts) > 1 else ""

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

# 简单记忆
memory = ConversationBufferMemory()
chain = ConversationChain(llm=llm, memory=memory, verbose=True)

response1 = chain.predict(input="我叫张三")
response2 = chain.predict(input="我叫什么名字？")  # 会记住张三

# 摘要记忆（适合长对话）
summary_memory = ConversationSummaryMemory(llm=llm)
chain = ConversationChain(llm=llm, memory=summary_memory)

from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings)

# 创建向量记忆
memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)

# 添加记忆
memory.save_context({"input": "我喜欢编程"}, {"output": "好的，我会记住你喜欢编程"})

# 使用记忆
relevant_memories = memory.load_memory_variables({"input": "我的爱好是什么？"})

from langchain.tools import DuckDuckGoSearchRun
from langchain.agents import initialize_agent

# 搜索工具
search = DuckDuckGoSearchRun()

# 创建搜索Agent
tools = [search]
agent = initialize_agent(
    tools,
    llm,
    agent="zero-shot-react-description",
    verbose=True
)

# 执行搜索
result = agent.invoke("最新的AI发展趋势是什么？")

from langchain_experimental.utilities import PythonREPL
from langchain.tools import Tool

# Python执行工具
python_repl = PythonREPL()
repl_tool = Tool(
    name="python_repl",
    description="执行Python代码",
    func=python_repl.run
)

# 创建代码Agent
agent = initialize_agent(
    [repl_tool],
    llm,
    agent="zero-shot-react-description",
    verbose=True
)

# 执行代码任务
result = agent.invoke("计算斐波那契数列前10项的和")

from langchain.agents import AgentExecutor
from langchain.schema import HumanMessage

class MultiAgentSystem:
    """多Agent系统"""
    
    def __init__(self):
        self.researcher = self._create_agent("研究员", "负责搜索和收集信息")
        self.writer = self._create_agent("写作者", "负责撰写和编辑内容")
        self.reviewer = self._create_agent("审核员", "负责审核和改进内容")
    
    def _create_agent(self, role, goal):
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"你是{role}，{goal}。"),
            ("human", "{input}")
        ])
        return prompt | llm
    
    def run(self, task):
        # 研究员收集信息
        research = self.researcher.invoke({"input": f"研究主题: {task}"})
        
        # 写作者撰写内容
        draft = self.writer.invoke({
            "input": f"基于以下研究内容撰写文章:\n{research.content}"
        })
        
        # 审核员审核
        review = self.reviewer.invoke({
            "input": f"审核并改进以下内容:\n{draft.content}"
        })
        
        return review.content