AI Agent技能每日速递：开发实战与工具链

AI Agent开发正在成为AI应用的核心竞争力。本文每日追踪AI Agent开发领域的最热技能和工具，涵盖Prompt工程、RAG技术、Agent框架等实战内容。

核心技能栈

1. Prompt工程高级技巧

结构化Prompt设计

class StructuredPrompt:
    """结构化Prompt模板"""
    
    @staticmethod
    def cot_prompting(question: str) -> str:
        """思维链Prompt"""
        return f"""
        让我们一步步思考这个问题：
        
        问题：{question}
        
        思考步骤：
        1. 理解问题：首先，我需要理解问题的核心是什么
        2. 分解问题：将复杂问题分解为子问题
        3. 逐步推理：对每个子问题进行逻辑推理
        4. 综合结论：将各部分推理整合得到最终答案
        
        请按照这个步骤详细思考并给出答案。
        """
    
    @staticmethod
    def few_shot_prompt(task: str, examples: list) -> str:
        """少样本学习Prompt"""
        prompt = f"任务：{task}\n\n示例：\n"
        
        for i, example in enumerate(examples, 1):
            prompt += f"\n示例{i}：\n"
            prompt += f"输入：{example['input']}\n"
            prompt += f"输出：{example['output']}\n"
        
        prompt += f"\n现在请处理以下输入：\n"
        return prompt
    
    @staticmethod
    def self_consistency_prompt(question: str, num_samples: int = 5) -> str:
        """自一致性Prompt"""
        return f"""
        问题：{question}
        
        请从不同角度思考这个问题，给出{num_samples}个可能的解决方案。
        然后选择最合理的一个，并说明理由。
        """

# 使用示例
prompt_engineer = StructuredPrompt()

# 复杂问题求解
complex_question = "如果要在月球建立永久基地，需要解决哪些关键问题？"
prompt = prompt_engineer.cot_prompting(complex_question)
print(prompt)
​

2. RAG系统构建

from typing import List, Dict, Any
import chromadb
from sentence_transformers import SentenceTransformer

class RAGSystem:
    """检索增强生成系统"""
    
    def __init__(self, embedding_model: str = "all-MiniLM-L6-v2"):
        self.embedding_model = SentenceTransformer(embedding_model)
        self.vector_store = chromadb.Client()
        self.collection = self.vector_store.create_collection(
            name="documents",
            metadata={"hnsw:space": "cosine"}
        )
    
    def add_documents(self, documents: List[Dict[str, Any]]):
        """添加文档到向量存储"""
        texts = [doc['text'] for doc in documents]
        embeddings = self.embedding_model.encode(texts)
        
        self.collection.add(
            embeddings=embeddings.tolist(),
            documents=texts,
            metadatas=[doc.get('metadata', {}) for doc in documents],
            ids=[doc.get('id', f"doc_{i}") for i, doc in enumerate(documents)]
        )
    
    def retrieve(self, query: str, top_k: int = 3) -> List[Dict[str, Any]]:
        """检索相关文档"""
        query_embedding = self.embedding_model.encode([query])
        
        results = self.collection.query(
            query_embeddings=query_embedding.tolist(),
            n_results=top_k
        )
        
        return [
            {
                'text': doc,
                'metadata': meta,
                'distance': dist
            }
            for doc, meta, dist in zip(
                results['documents'][0],
                results['metadatas'][0],
                results['distances'][0]
            )
        ]
    
    def generate(self, query: str, context: List[str]) -> str:
        """基于检索上下文生成回答"""
        prompt = f"""
        基于以下参考信息回答问题：
        
        参考信息：
        {' '.join(context)}
        
        问题：{query}
        
        请提供准确、详细的回答。
        """
        
        # 调用LLM生成
        response = self.llm_client.generate(prompt)
        return response

# 高级RAG：混合检索
class AdvancedRAG(RAGSystem):
    """高级RAG系统"""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.keyword_index = {}  # 关键词索引
    
    def hybrid_retrieve(self, query: str, alpha: float = 0.5, top_k: int = 5):
        """混合检索：向量检索 + 关键词检索"""
        # 向量检索
        vector_results = self.retrieve(query, top_k)
        
        # 关键词检索
        keyword_results = self.keyword_search(query, top_k)
        
        # 结果融合（Reciprocal Rank Fusion）
        fused_results = self.reciprocal_rank_fusion(
            vector_results, 
            keyword_results, 
            alpha
        )
        
        return fused_results[:top_k]
    
    def reciprocal_rank_fusion(self, results1: List, results2: List, k: int = 60):
        """RRF算法融合结果"""
        scores = {}
        
        for rank, doc in enumerate(results1):
            doc_id = doc.get('id', '')
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
        
        for rank, doc in enumerate(results2):
            doc_id = doc.get('id', '')
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
        
        # 按分数排序
        ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return ranked
​

3. Agent框架实战

LangChain Agent

from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain.llms import OpenAI
from langchain.memory import ConversationBufferMemory

class LangChainAgent:
    """LangChain Agent实现"""
    
    def __init__(self, openai_api_key: str):
        self.llm = OpenAI(temperature=0, openai_api_key=openai_api_key)
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )
    
    def create_tools(self):
        """创建Agent工具集"""
        tools = [
            Tool(
                name="Search",
                func=self.search_internet,
                description="在互联网上搜索最新信息"
            ),
            Tool(
                name="Calculator",
                func=self.calculate,
                description="执行数学计算"
            ),
            Tool(
                name="CodeExecutor",
                func=self.execute_code,
                description="执行Python代码"
            )
        ]
        return tools
    
    def build_agent(self):
        """构建Agent"""
        tools = self.create_tools()
        
        # ReAct Prompt模板
        template = """Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought: {agent_scratchpad}"""
        
        agent = create_react_agent(
            self.llm,
            tools,
            prompt=template
        )
        
        agent_executor = AgentExecutor(
            agent=agent,
            tools=tools,
            verbose=True,
            memory=self.memory,
            max_iterations=5
        )
        
        return agent_executor
    
    @staticmethod
    def search_internet(query: str) -> str:
        """搜索工具"""
        # 实际应用中调用搜索API
        return f"搜索结果：关于'{query}'的信息..."
    
    @staticmethod
    def calculate(expression: str) -> str:
        """计算工具"""
        try:
            result = eval(expression)
            return f"计算结果：{result}"
        except:
            return "计算错误"
    
    @staticmethod
    def execute_code(code: str) -> str:
        """代码执行工具"""
        try:
            exec_globals = {}
            exec(code, exec_globals)
            return "代码执行成功"
        except Exception as e:
            return f"执行错误：{str(e)}"

# 使用示例
agent = LangChainAgent("your-api-key")
agent_executor = agent.build_agent()

result = agent_executor.invoke({
    "input": "计算23*45+67的结果"
})
print(result['output'])
​

AutoGen多Agent协作

import autogen

class AutoGENTeam:
    """AutoGen多Agent协作"""
    
    def __init__(self, config_list: list):
        self.config_list = config_list
    
    def create_assistant_agent(self, name: str, system_message: str):
        """创建助手Agent"""
        return autogen.AssistantAgent(
            name=name,
            llm_config={
                "config_list": self.config_list,
                "temperature": 0.7
            },
            system_message=system_message
        )
    
    def create_user_proxy(self):
        """创建用户代理"""
        return autogen.UserProxyAgent(
            name="user_proxy",
            human_input_mode="NEVER",
            max_consecutive_auto_reply=10,
            is_termination_msg=lambda x: x.get("content", "").find("TERMINATE") >= 0,
            code_execution_config={
                "work_dir": "coding",
                "use_docker": False
            }
        )
    
    def build_team(self):
        """构建Agent团队"""
        # 定义不同角色的Agent
        product_manager = self.create_assistant_agent(
            "product_manager",
            "你是产品经理，负责需求分析和产品规划。"
        )
        
        developer = self.create_assistant_agent(
            "developer",
            "你是开发工程师，负责编写代码和技术实现。"
        )
        
        code_reviewer = self.create_assistant_agent(
            "code_reviewer",
            "你是代码审查专家，负责代码质量和最佳实践。"
        )
        
        user_proxy = self.create_user_proxy()
        
        return {
            'product_manager': product_manager,
            'developer': developer,
            'code_reviewer': code_reviewer,
            'user_proxy': user_proxy
        }
    
    def start_collaboration(self, task: str):
        """启动协作"""
        team = self.build_team()
        
        # 定义协作流程
        user_proxy = team['user_proxy']
        product_manager = team['product_manager']
        developer = team['developer']
        code_reviewer = team['code_reviewer']
        
        # 开始协作
        user_proxy.initiate_chat(
            product_manager,
            message=f"任务：{task}\n请开始需求分析。"
        )
        
        # 产品经理与开发者对话
        product_manager.initiate_chat(
            developer,
            message=product_manager.last_message()["content"]
        )
        
        # 开发者与代码审查者对话
        developer.initiate_chat(
            code_reviewer,
            message=developer.last_message()["content"]
        )
        
        return {
            'requirements': product_manager.last_message(),
            'code': developer.last_message(),
            'review': code_reviewer.last_message()
        }

# 使用示例
config_list = [
    {
        'model': 'gpt-4',
        'api_key': 'your-api-key'
    }
]

team = AutoGENTeam(config_list)
result = team.start_collaboration("开发一个待办事项管理应用")
​

4. 记忆系统设计

from typing import Dict, List, Any
from datetime import datetime, timedelta
import json

class AgentMemory:
    """Agent记忆系统"""
    
    def __init__(self):
        self.short_term = []  # 短期记忆
        self.long_term = {}   # 长期记忆
        self.working_memory = {}  # 工作记忆
    
    def add_experience(self, experience: Dict[str, Any]):
        """添加经验"""
        timestamp = datetime.now()
        
        experience['timestamp'] = timestamp
        experience['importance'] = self.calculate_importance(experience)
        
        # 添加到短期记忆
        self.short_term.append(experience)
        
        # 如果重要性高，转移到长期记忆
        if experience['importance'] > 0.7:
            self.consolidate_memory(experience)
        
        # 维持短期记忆容量
        if len(self.short_term) > 10:
            self.short_term.pop(0)
    
    def calculate_importance(self, experience: Dict) -> float:
        """计算记忆重要性"""
        importance = 0.0
        
        # 新近性
        hours_ago = (datetime.now() - experience['timestamp']).total_seconds() / 3600
        recency = 1.0 / (1.0 + hours_ago)
        importance += recency * 0.3
        
        # 情感强度
        emotion = experience.get('emotion', 0)
        importance += abs(emotion) * 0.4
        
        # 访问频率
        access_count = experience.get('access_count', 0)
        importance += min(access_count * 0.1, 0.3)
        
        return min(importance, 1.0)
    
    def consolidate_memory(self, experience: Dict):
        """记忆固化"""
        key = f"{experience['timestamp']}_memory"
        self.long_term[key] = experience
    
    def retrieve_relevant(self, query: str, top_k: int = 3) -> List[Dict]:
        """检索相关记忆"""
        all_memories = self.short_term + list(self.long_term.values())
        
        # 简单的关键词匹配
        relevant = []
        for memory in all_memories:
            if query.lower() in str(memory).lower():
                relevant.append(memory)
        
        # 按重要性排序
        relevant.sort(key=lambda x: x['importance'], reverse=True)
        
        return relevant[:top_k]

# 向量记忆存储
class VectorMemory:
    """基于向量的记忆存储"""
    
    def __init__(self, embedding_model):
        self.embedding_model = embedding_model
        self.memories = []
        self.embeddings = []
    
    def add_memory(self, memory: Dict):
        """添加记忆"""
        text = memory.get('text', '')
        embedding = self.embedding_model.encode(text)
        
        self.memories.append(memory)
        self.embeddings.append(embedding)
    
    def search(self, query: str, top_k: int = 5) -> List[Dict]:
        """搜索相似记忆"""
        from sklearn.metrics.pairwise import cosine_similarity
        
        query_embedding = self.embedding_model.encode([query])
        similarities = cosine_similarity(query_embedding, self.embeddings)[0]
        
        # 获取top-k
        top_indices = similarities.argsort()[-top_k:][::-1]
        
        results = [
            {
                'memory': self.memories[i],
                'similarity': similarities[i]
            }
            for i in top_indices
        ]
        
        return results
​

工具链推荐

开发框架

1. LangChain：全功能AI应用框架
2. AutoGen：多Agent协作框架
3. CrewAI：角色扮演Agent系统
4. Semantic Kernel：微软轻量级框架

向量数据库

1. Chroma：轻量级本地向量数据库
2. Pinecone：托管向量数据库服务
3. Weaviate：开源向量搜索引擎
4. Qdrant：高性能向量数据库

监控调试

1. LangSmith：LangChain调试平台
2. Weights & Biases：实验跟踪
3. MLflow：机器学习生命周期管理

最佳实践

1. Agent设计原则

# 设计可测试的Agent
class TestableAgent:
    def __init__(self, tools, llm):
        self.tools = tools
        self.llm = llm
    
    def process(self, input_data):
        # 明确的输入输出
        try:
            # 可观察的步骤
            step1_result = self.step1(input_data)
            step2_result = self.step2(step1_result)
            final_result = self.step3(step2_result)
            
            return {
                'success': True,
                'result': final_result,
                'intermediate_steps': [step1_result, step2_result]
            }
        except Exception as e:
            return {
                'success': False,
                'error': str(e),
                'context': input_data
            }
​

2. 性能优化

# 缓存常见查询
from functools import lru_cache

class OptimizedAgent:
    @lru_cache(maxsize=100)
    def cached_llm_call(self, prompt):
        return self.llm.generate(prompt)
    
    # 批量处理
    def batch_process(self, inputs: List[str]):
        return self.llm.generate_batch(inputs)
​

3. 错误处理

class RobustAgent:
    def safe_execute(self, tool_func, *args, **kwargs):
        """安全执行工具"""
        max_retries = 3
        
        for attempt in range(max_retries):
            try:
                return tool_func(*args, **kwargs)
            except Exception as e:
                if attempt == max_retries - 1:
                    # 最后一次尝试失败，返回错误
                    return {
                        'error': str(e),
                        'failed_attempts': max_retries
                    }
                else:
                    # 重试
                    continue
​

总结

AI Agent开发需要掌握：

1. Prompt工程：有效与LLM沟通
2. RAG技术：增强知识检索能力
3. Agent框架：快速构建智能体
4. 记忆系统：持久化和检索经验
5. 工具使用：扩展Agent能力

持续关注Agent技术的快速发展，不断学习和实践，是成为AI Agent开发专家的关键。