LLM应用开发——Agent智能体

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
import json
import re

# 定义工具
@tool
def search_weather(city: str) -> str:
    """查询城市天气信息"""
    weather_data = {
        "北京": "晴，温度8-15°C，空气质量良好",
        "上海": "多云，温度12-18°C，湿度80%",
        "深圳": "阴，温度20-25°C，有阵雨"
    }
    return weather_data.get(city, f"未找到{city}的天气信息")

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

@tool
def search_web(query: str) -> str:
    """搜索网络信息"""
    return f"关于'{query}'的搜索结果：这是一个模拟的搜索结果。"

class SimpleReActAgent:
    def __init__(self, tools: list):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.tools = {t.name: t for t in tools}
        self.tools_description = "\n".join(
            f"- {t.name}: {t.description}"
            for t in tools
        )
        
        self.prompt = ChatPromptTemplate.from_template("""
你是一个能够使用工具的AI助手。请使用ReAct格式回答问题。

可用工具：
{tools}

使用格式：
Thought: 思考当前需要做什么
Action: 工具名称
Action Input: 工具输入（JSON格式）
Observation: 工具返回结果（由系统填充）
... (可以重复多次)
Thought: 得出最终结论
Final Answer: 最终答案

问题：{question}

{agent_scratchpad}
""")
    
    def run(self, question: str, max_iterations: int = 5) -> str:
        """运行Agent"""
        scratchpad = ""
        
        for i in range(max_iterations):
            # 构建提示
            prompt_value = self.prompt.format(
                tools=self.tools_description,
                question=question,
                agent_scratchpad=scratchpad
            )
            
            # 获取LLM响应
            response = self.llm.invoke(prompt_value).content
            
            # 检查是否得出最终答案
            if "Final Answer:" in response:
                final_answer = response.split("Final Answer:")[-1].strip()
                return final_answer
            
            # 解析并执行Action
            action_match = re.search(
                r'Action:\s*(\w+)\s*\nAction Input:\s*(.+?)(?=\n|$)',
                response, re.DOTALL
            )
            
            if action_match:
                action_name = action_match.group(1)
                action_input = action_match.group(2).strip()
                
                # 执行工具
                if action_name in self.tools:
                    try:
                        # 解析JSON输入
                        if action_input.startswith('{'):
                            input_dict = json.loads(action_input)
                            result = self.tools[action_name].invoke(input_dict)
                        else:
                            result = self.tools[action_name].invoke(action_input)
                    except Exception as e:
                        result = f"工具执行错误: {e}"
                else:
                    result = f"未找到工具: {action_name}"
                
                # 更新scratchpad
                scratchpad += f"{response}\nObservation: {result}\n"
            else:
                scratchpad += response + "\n"
        
        return "达到最大迭代次数，未能得出答案"

# 使用示例
agent = SimpleReActAgent([search_weather, calculate, search_web])
answer = agent.run("北京今天天气怎么样？如果温度是15度，加上5度是多少？")
print(answer)

class TaskPlanner:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        
        self.plan_prompt = ChatPromptTemplate.from_template("""
你是一个任务规划专家。请将用户的复杂目标分解为具体的执行步骤。

目标：{goal}

请以JSON格式输出任务分解：
goal_summary
    ],
    "success_criteria": "成功标准"
}}

请确保步骤之间的依赖关系正确，步骤要具体可执行。
""")
    
    def plan(self, goal: str) -> dict:
        """生成执行计划"""
        prompt = self.plan_prompt.format(goal=goal)
        response = self.llm.invoke(prompt).content
        
        # 提取JSON
        import json
        json_match = re.search(r'\{[\s\S]*\}', response)
        if json_match:
            return json.loads(json_match.group())
        return {"error": "无法解析计划"}

# 使用示例
planner = TaskPlanner()
plan = planner.plan("帮我调研Python Web框架，比较Flask、FastAPI和Django，并给出选型建议")
print(json.dumps(plan, ensure_ascii=False, indent=2))

from typing import List, Dict, Any

class PlanAndExecuteAgent:
    def __init__(self, tools: list):
        self.planner = TaskPlanner()
        self.executor = SimpleReActAgent(tools)
        self.llm = ChatOpenAI(model="gpt-4o")
    
    def run(self, goal: str) -> str:
        """规划并执行"""
        # 1. 生成计划
        print("📋 正在规划任务...")
        plan = self.planner.plan(goal)
        
        if "error" in plan:
            return f"规划失败: {plan['error']}"
        
        print(f"✅ 计划包含 {len(plan['steps'])} 个步骤")
        
        # 2. 按顺序执行步骤
        results = []
        for step in plan['steps']:
            print(f"\n🔄 执行步骤 {step['id']}: {step['description']}")
            
            # 执行单个步骤
            step_result = self.executor.run(step['description'])
            results.append({
                "step_id": step['id'],
                "description": step['description'],
                "result": step_result
            })
            
            print(f"✅ 步骤完成")
        
        # 3. 汇总结果
        summary = self._summarize_results(goal, results)
        return summary
    
    def _summarize_results(self, goal: str, results: List[Dict]) -> str:
        """汇总执行结果"""
        results_text = "\n".join(
            f"步骤{r['step_id']}: {r['description']}\n结果: {r['result']}"
            for r in results
        )
        
        summary_prompt = f"""
原始目标：{goal}

执行结果：
{results_text}

请根据以上执行结果，给出最终的综合回答。
"""
        
        response = self.llm.invoke(summary_prompt)
        return response.content

from collections import deque
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class MemoryItem:
    content: str
    type: str  # observation, thought, action, result
    timestamp: datetime
    importance: float = 1.0

class ShortTermMemory:
    def __init__(self, capacity: int = 20):
        self.memory = deque(maxlen=capacity)
    
    def add(self, content: str, type: str, importance: float = 1.0):
        """添加记忆"""
        item = MemoryItem(
            content=content,
            type=type,
            timestamp=datetime.now(),
            importance=importance
        )
        self.memory.append(item)
    
    def get_recent(self, n: int = 5) -> List[MemoryItem]:
        """获取最近n条记忆"""
        return list(self.memory)[-n:]
    
    def get_by_type(self, type: str) -> List[MemoryItem]:
        """按类型获取记忆"""
        return [m for m in self.memory if m.type == type]
    
    def to_context(self) -> str:
        """转换为上下文字符串"""
        return "\n".join(
            f"[{m.type}] {m.content}"
            for m in self.memory
        )

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document

class LongTermMemory:
    def __init__(self, persist_dir: str = "./agent_memory"):
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = Chroma(
            collection_name="agent_memory",
            embedding_function=self.embeddings,
            persist_directory=persist_dir
        )
    
    def add(self, content: str, metadata: dict = None):
        """存储记忆"""
        doc = Document(
            page_content=content,
            metadata=metadata or {}
        )
        self.vectorstore.add_documents([doc])
    
    def search(self, query: str, k: int = 5) -> List[Document]:
        """检索相关记忆"""
        return self.vectorstore.similarity_search(query, k=k)
    
    def get_relevant_context(self, query: str, k: int = 3) -> str:
        """获取相关上下文"""
        docs = self.search(query, k=k)
        return "\n".join(doc.page_content for doc in docs)

class ReflectionMemory:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o")
        self.experiences = []
    
    def add_experience(self, task: str, actions: list, outcome: str, success: bool):
        """记录经验"""
        self.experiences.append({
            "task": task,
            "actions": actions,
            "outcome": outcome,
            "success": success,
            "timestamp": datetime.now()
        })
    
    def reflect(self) -> str:
        """反思近期经验"""
        if not self.experiences:
            return "暂无经验可供反思"
        
        recent = self.experiences[-5:]
        experiences_text = "\n".join(
            f"任务: {e['task']}\n行动: {e['actions']}\n结果: {e['outcome']}\n成功: {e['success']}"
            for e in recent
        )
        
        prompt = f"""
请分析以下执行经验，总结规律和教训：

{experiences_text}

请给出：
1. 成功模式：哪些做法有效
2. 失败原因：哪些情况导致失败
3. 改进建议：如何提高成功率
"""
        
        response = self.llm.invoke(prompt)
        return response.content

from typing import Callable
from enum import Enum

class AgentRole(Enum):
    COORDINATOR = "coordinator"
    RESEARCHER = "researcher"
    CODER = "coder"
    REVIEWER = "reviewer"

class SpecializedAgent:
    def __init__(self, role: AgentRole, tools: list = None):
        self.role = role
        self.llm = ChatOpenAI(model="gpt-4o")
        self.tools = tools or []
        
        self.system_prompts = {
            AgentRole.COORDINATOR: """
你是团队协调者，负责：
1. 理解任务需求
2. 分配任务给合适的团队成员
3. 整合各方结果
4. 做出最终决策
""",
            AgentRole.RESEARCHER: """
你是研究专家，负责：
1. 收集和分析信息
2. 调研最佳实践
3. 评估技术方案
4. 提供研究报告
""",
            AgentRole.CODER: """
你是编程专家，负责：
1. 编写高质量代码
2. 实现技术方案
3. 处理技术细节
4. 确保代码可运行
""",
            AgentRole.REVIEWER: """
你是审查专家，负责：
1. 审查代码质量
2. 发现潜在问题
3. 提供改进建议
4. 确保符合标准
"""
        }
    
    def process(self, task: str, context: str = "") -> str:
        """处理任务"""
        prompt = f"""
{self.system_prompts[self.role]}

当前上下文：
{context}

任务：
{task}

请完成任务并给出详细结果。
"""
        response = self.llm.invoke(prompt)
        return response.content

class MultiAgentSystem:
    def __init__(self):
        self.agents = {
            AgentRole.COORDINATOR: SpecializedAgent(AgentRole.COORDINATOR),
            AgentRole.RESEARCHER: SpecializedAgent(AgentRole.RESEARCHER),
            AgentRole.CODER: SpecializedAgent(AgentRole.CODER),
            AgentRole.REVIEWER: SpecializedAgent(AgentRole.REVIEWER),
        }
        self.conversation_history = []
    
    def run(self, goal: str) -> str:
        """运行多Agent协作"""
        context = ""
        
        # 1. 协调者分析任务
        print("🎯 协调者分析任务...")
        coordinator_analysis = self.agents[AgentRole.COORDINATOR].process(
            f"分析以下任务，确定需要哪些专家参与，以及执行顺序：\n{goal}"
        )
        context += f"协调者分析：\n{coordinator_analysis}\n\n"
        
        # 2. 研究者调研
        print("🔍 研究者调研...")
        research_result = self.agents[AgentRole.RESEARCHER].process(
            goal, context
        )
        context += f"研究结果：\n{research_result}\n\n"
        
        # 3. 编码者实现
        print("💻 编码者实现...")
        code_result = self.agents[AgentRole.CODER].process(
            goal, context
        )
        context += f"代码实现：\n{code_result}\n\n"
        
        # 4. 审查者审查
        print("🔎 审查者审查...")
        review_result = self.agents[AgentRole.REVIEWER].process(
            f"审查以下实现：\n{code_result}", context
        )
        context += f"审查结果：\n{review_result}\n\n"
        
        # 5. 协调者汇总
        print("📋 协调者汇总...")
        final_result = self.agents[AgentRole.COORDINATOR].process(
            f"汇总团队工作结果，给出最终交付物：\n目标：{goal}",
            context
        )
        
        return final_result

# 使用示例
mas = MultiAgentSystem()
result = mas.run("实现一个Python函数，用于解析CSV文件并计算每列的统计信息")
print(result)

class DynamicToolSelector:
    def __init__(self, tools: list):
        self.tools = {t.name: t for t in tools}
        self.llm = ChatOpenAI(model="gpt-4o")
    
    def select_tools(self, task: str) -> list:
        """根据任务动态选择工具"""
        tools_info = "\n".join(
            f"- {name}: {tool.description}"
            for name, tool in self.tools.items()
        )
        
        prompt = f"""
任务：{task}

可用工具：
{tools_info}

请选择完成此任务所需的工具（返回工具名列表，JSON格式）：
"""
        
        response = self.llm.invoke(prompt).content
        selected = json.loads(response)
        
        return [self.tools[name] for name in selected if name in self.tools]

class ToolChain:
    def __init__(self, tools: list):
        self.tools = {t.name: t for t in tools}
    
    def execute_chain(self, chain: List[Dict]) -> List[Dict]:
        """执行工具链
        
        chain格式:
        [
            {"tool": "search", "input": "query", "output_var": "search_result"},
            {"tool": "summarize", "input": "${search_result}", "output_var": "summary"}
        ]
        """
        variables = {}
        results = []
        
        for step in chain:
            tool_name = step["tool"]
            input_val = step["input"]
            
            # 替换变量引用
            if input_val.startswith("${") and input_val.endswith("}"):
                var_name = input_val[2:-1]
                input_val = variables.get(var_name, input_val)
            
            # 执行工具
            if tool_name in self.tools:
                output = self.tools[tool_name].invoke(input_val)
                variables[step["output_var"]] = output
                results.append({
                    "step": step,
                    "output": output
                })
            else:
                results.append({
                    "step": step,
                    "error": f"Tool {tool_name} not found"
                })
        
        return results

class RobustToolExecutor:
    def __init__(self, tools: list, max_retries: int = 3):
        self.tools = {t.name: t for t in tools}
        self.max_retries = max_retries
        self.llm = ChatOpenAI(model="gpt-4o")
    
    def execute(self, tool_name: str, input_data: Any) -> Dict:
        """带重试的工具执行"""
        errors = []
        
        for attempt in range(self.max_retries):
            try:
                if tool_name not in self.tools:
                    # 尝试找到相似工具
                    suggested = self._suggest_tool(tool_name)
                    if suggested:
                        tool_name = suggested
                    else:
                        return {"error": f"工具 {tool_name} 不存在"}
                
                result = self.tools[tool_name].invoke(input_data)
                return {"success": True, "result": result}
                
            except Exception as e:
                errors.append(str(e))
                
                # 尝试修复输入
                if attempt < self.max_retries - 1:
                    input_data = self._fix_input(tool_name, input_data, str(e))
        
        return {
            "error": "执行失败",
            "attempts": errors
        }
    
    def _suggest_tool(self, tool_name: str) -> Optional[str]:
        """建议相似工具"""
        from difflib import get_close_matches
        matches = get_close_matches(
            tool_name, 
            self.tools.keys(), 
            n=1, 
            cutoff=0.6
        )
        return matches[0] if matches else None
    
    def _fix_input(self, tool_name: str, input_data: Any, error: str) -> Any:
        """尝试修复输入"""
        prompt = f"""
工具 {tool_name} 执行失败。
输入: {input_data}
错误: {error}

请修正输入格式（只返回修正后的输入）：
"""
        response = self.llm.invoke(prompt)
        return response.content.strip()

from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchResults
from langchain_core.tools import tool
import json

# 定义工具
@tool
def web_search(query: str) -> str:
    """搜索互联网获取信息"""
    search = DuckDuckGoSearchResults()
    return search.run(query)

@tool
def take_notes(content: str) -> str:
    """记录研究笔记"""
    # 实际应用中可以保存到文件或数据库
    return f"已记录笔记: {content[:100]}..."

@tool
def write_report(topic: str, notes: str) -> str:
    """根据笔记生成研究报告"""
    llm = ChatOpenAI(model="gpt-4o")
    prompt = f"""
基于以下研究笔记，为主题"{topic}"生成一份结构化的研究报告。

研究笔记：
{notes}

报告应包含：
1. 摘要
2. 背景介绍
3. 主要发现
4. 结论与建议
"""
    response = llm.invoke(prompt)
    return response.content

class ResearchAssistant:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.tools = [web_search, take_notes, write_report]
        self.notes = []
        
        self.system_prompt = """
你是一个专业的研究助手Agent。你的任务是帮助用户进行深入研究。

工作流程：
1. 理解研究主题和目标
2. 制定研究计划
3. 使用搜索工具收集信息
4. 整理和记录重要发现
5. 生成研究报告

可用工具：
- web_search: 搜索互联网信息
- take_notes: 记录研究笔记
- write_report: 生成研究报告

请根据用户需求，逐步完成研究任务。
"""
    
    def research(self, topic: str) -> str:
        """执行研究任务"""
        print(f"🔬 开始研究: {topic}\n")
        
        # 1. 制定研究计划
        plan_prompt = f"""
研究主题: {topic}

请制定一个详细的研究计划，包括：
1. 需要搜索的关键问题（3-5个）
2. 预期的研究产出
"""
        plan = self.llm.invoke(plan_prompt).content
        print(f"📋 研究计划:\n{plan}\n")
        
        # 2. 执行搜索
        search_queries = self._extract_queries(plan)
        all_findings = []
        
        for query in search_queries:
            print(f"🔍 搜索: {query}")
            results = web_search.invoke(query)
            all_findings.append(f"查询: {query}\n结果: {results}")
        
        # 3. 整理笔记
        notes_content = "\n\n".join(all_findings)
        take_notes.invoke(notes_content)
        
        # 4. 生成报告
        print("\n📝 生成研究报告...")
        report = write_report.invoke({"topic": topic, "notes": notes_content})
        
        return report
    
    def _extract_queries(self, plan: str) -> List[str]:
        """从计划中提取搜索查询"""
        prompt = f"""
从以下研究计划中提取关键搜索查询，返回JSON列表：

{plan}

返回格式：["查询1", "查询2", ...]
"""
        response = self.llm.invoke(prompt).content
        try:
            queries = json.loads(response)
            return queries[:5]  # 限制最多5个查询
        except:
            return [plan.split('\n')[0]]  # 回退到主题

# 使用示例
assistant = ResearchAssistant()
report = assistant.research("2024年大语言模型的最新发展趋势")
print(report)