关于大模型应用开发的几点记录

import openai

# 初始化客户端
client = openai.OpenAI(api_key="your-api-key")

# 基础调用
response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello, how are you?"}
    ]
)

print(response.choices[0].message.content)

def stream_chat(messages):
    stream = client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        stream=True
    )
    
    for chunk in stream:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

# 使用
for content in stream_chat([
    {"role": "user", "content": "Tell me a story"}
]):
    print(content, end="", flush=True)

import json

# 定义函数
def get_weather(location, unit="celsius"):
    """获取天气信息"""
    # 实际调用天气 API
    return {
        "location": location,
        "temperature": 25,
        "condition": "sunny",
        "unit": unit
    }

# 函数描述
functions = [
    {
        "name": "get_weather",
        "description": "获取指定地点的天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "城市名称"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度单位"
                }
            },
            "required": ["location"]
        }
    }
]

# 使用函数调用
response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "user", "content": "北京现在的天气怎么样？"}
    ],
    functions=functions,
    function_call="auto"
)

message = response.choices[0].message

if message.function_call:
    function_name = message.function_call.name
    function_args = json.loads(message.function_call.arguments)
    
    # 执行函数
    if function_name == "get_weather":
        result = get_weather(**function_args)
        
        # 将结果返回给模型
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "user", "content": "北京现在的天气怎么样？"},
                message,
                {"role": "function", "name": function_name, "content": json.dumps(result)}
            ]
        )
        
        print(response.choices[0].message.content)

def structured_prompt(prompt_type, **kwargs):
    prompts = {
        "qa": f"""
Question: {kwargs['question']}
Context: {kwargs.get('context', '')}

Answer the question based on the context. If the answer is not in the context, say "I don't know".
""",
        "summarize": f"""
Text: {kwargs['text']}

Summarize the above text in 2-3 sentences.
""",
        "extract": f"""
Text: {kwargs['text']}
Information to extract: {kwargs['target']}

Extract the following information from the text.
"""
    }
    
    return prompts.get(prompt_type, "")

# 使用
messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": structured_prompt("qa", 
        question="What is the capital of France?",
        context="Paris is the capital and largest city of France."
    )}
]

def chain_of_thought(prompt):
    messages = [
        {
            "role": "system",
            "content": """You are a helpful assistant. Think step by step:
1. Understand the problem
2. Identify the relevant information
3. Reason through the problem
4. Provide the final answer
"""
        },
        {"role": "user", "content": prompt}
    ]
    
    return client.chat.completions.create(
        model="gpt-4",
        messages=messages
    ).choices[0].message.content

# 使用
result = chain_of_thought(
    "A store sells apples for $2 each and oranges for $1.5 each. "
    "If someone buys 3 apples and 4 oranges, how much do they spend?"
)
print(result)

def few_shot_prompt(examples, query):
    prompt = "Here are some examples:\n\n"
    
    for example in examples:
        prompt += f"Input: {example['input']}\n"
        prompt += f"Output: {example['output']}\n\n"
    
    prompt += f"Input: {query}\n"
    prompt += "Output:"
    
    return prompt

# 使用
examples = [
    {"input": "cat", "output": "meow"},
    {"input": "dog", "output": "bark"},
    {"input": "bird", "output": "chirp"}
]

prompt = few_shot_prompt(examples, "cow")

response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "user", "content": prompt}
    ]
)

print(response.choices[0].message.content)

from typing import List, Dict
import openai

class RAGSystem:
    def __init__(self, documents: List[str]):
        self.documents = documents
        self.embeddings = self._create_embeddings()
    
    def _create_embeddings(self):
        # 为所有文档创建嵌入
        embeddings = []
        for doc in self.documents:
            response = openai.Embedding.create(
                model="text-embedding-ada-002",
                input=doc
            )
            embeddings.append(response['data'][0]['embedding'])
        return embeddings
    
    def search(self, query: str, top_k: int = 3) -> List[Dict[str, str]]:
        # 为查询创建嵌入
        query_response = openai.Embedding.create(
            model="text-embedding-ada-002",
            input=query
        )
        query_embedding = query_response['data'][0]['embedding']
        
        # 计算相似度
        similarities = []
        for i, doc_embedding in enumerate(self.embeddings):
            similarity = self._cosine_similarity(query_embedding, doc_embedding)
            similarities.append({
                'document': self.documents[i],
                'similarity': similarity
            })
        
        # 返回最相似的文档
        similarities.sort(key=lambda x: x['similarity'], reverse=True)
        return similarities[:top_k]
    
    def _cosine_similarity(self, a, b):
        # 计算余弦相似度
        dot_product = sum(x * y for x, y in zip(a, b))
        magnitude_a = sum(x ** 2 for x in a) ** 0.5
        magnitude_b = sum(y ** 2 for y in b) ** 0.5
        return dot_product / (magnitude_a * magnitude_b)
    
    def generate_answer(self, query: str) -> str:
        # 检索相关文档
        context_docs = self.search(query)
        context = "\n".join([doc['document'] for doc in context_docs])
        
        # 生成回答
        messages = [
            {
                "role": "system",
                "content": "You are a helpful assistant. Answer the question based on the provided context."
            },
            {
                "role": "user",
                "content": f"""Context: {context}

Question: {query}

Answer:"""
            }
        ]
        
        response = client.chat.completions.create(
            model="gpt-4",
            messages=messages
        )
        
        return response.choices[0].message.content

# 使用
documents = [
    "Python is a high-level programming language.",
    "JavaScript is used for web development.",
    "Python is widely used in data science and machine learning.",
    "JavaScript can run in both browsers and servers."
]

rag = RAGSystem(documents)
answer = rag.generate_answer("What is Python used for?")
print(answer)

import chromadb
from chromadb.config import Settings

class VectorRAGSystem:
    def __init__(self):
        self.client = chromadb.Client(Settings(
            chroma_db_impl="duckdb+parquet",
            persist_directory="./chroma_db"
        ))
        self.collection = self.client.get_or_create_collection("documents")
    
    def add_documents(self, documents: List[str], metadatas: List[Dict] = None):
        # 为文档创建嵌入并存储
        embeddings = []
        for doc in documents:
            response = openai.Embedding.create(
                model="text-embedding-ada-002",
                input=doc
            )
            embeddings.append(response['data'][0]['embedding'])
        
        # 添加到向量数据库
        self.collection.add(
            documents=documents,
            embeddings=embeddings,
            metadatas=metadatas or [{}] * len(documents),
            ids=[f"doc_{i}" for i in range(len(documents))]
        )
    
    def search(self, query: str, n_results: int = 3) -> List[Dict]:
        # 为查询创建嵌入
        query_response = openai.Embedding.create(
            model="text-embedding-ada-002",
            input=query
        )
        query_embedding = query_response['data'][0]['embedding']
        
        # 搜索向量数据库
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results
        )
        
        return results['documents'][0]
    
    def generate_answer(self, query: str) -> str:
        # 检索相关文档
        context_docs = self.search(query)
        context = "\n".join(context_docs)
        
        # 生成回答
        messages = [
            {
                "role": "system",
                "content": "You are a helpful assistant. Answer the question based on the provided context."
            },
            {
                "role": "user",
                "content": f"""Context: {context}

Question: {query}

Answer:"""
            }
        ]
        
        response = client.chat.completions.create(
            model="gpt-4",
            messages=messages
        )
        
        return response.choices[0].message.content

# 使用
rag = VectorRAGSystem()

# 添加文档
documents = [
    {"text": "Python is a high-level programming language.", "source": "doc1"},
    {"text": "JavaScript is used for web development.", "source": "doc2"},
    {"text": "Python is widely used in data science.", "source": "doc3"}
]

rag.add_documents(
    [doc["text"] for doc in documents],
    [{"source": doc["source"]} for doc in documents]
)

# 生成回答
answer = rag.generate_answer("What is Python used for?")
print(answer)

class Agent:
    def __init__(self, name: str, tools: List[callable]):
        self.name = name
        self.tools = tools
        self.client = openai.OpenAI()
    
    def run(self, task: str) -> str:
        # 创建工具描述
        tool_descriptions = []
        for tool in self.tools:
            tool_descriptions.append({
                "name": tool.__name__,
                "description": tool.__doc__,
                "parameters": tool.__annotations__
            })
        
        # 调用模型
        messages = [
            {
                "role": "system",
                "content": f"""You are {self.name}. You have access to the following tools:
{', '.join([tool['name'] for tool in tool_descriptions])}

When you need to use a tool, respond with the tool name and arguments in the format:
TOOL: tool_name
ARGUMENTS: {{key: value}}"""
            },
            {"role": "user", "content": task}
        ]
        
        while True:
            response = self.client.chat.completions.create(
                model="gpt-4",
                messages=messages
            )
            
            result = response.choices[0].message.content
            
            # 检查是否需要使用工具
            if result.startswith("TOOL:"):
                # 解析工具调用
                lines = result.split("\n")
                tool_name = lines[0].replace("TOOL:", "").strip()
                arguments = json.loads(lines[1].replace("ARGUMENTS:", "").strip())
                
                # 执行工具
                tool = next((t for t in self.tools if t.__name__ == tool_name), None)
                if tool:
                    tool_result = tool(**arguments)
                    
                    # 将工具结果添加到消息中
                    messages.append({"role": "assistant", "content": result})
                    messages.append({"role": "user", "content": f"Tool result: {tool_result}"})
                else:
                    messages.append({"role": "assistant", "content": f"Unknown tool: {tool_name}"})
            else:
                return result

# 定义工具
def search_web(query: str) -> str:
    """Search the web for information"""
    # 实际实现会调用搜索 API
    return f"Search results for: {query}"

def get_current_time() -> str:
    """Get the current time"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 创建 Agent
agent = Agent("Researcher", [search_web, get_current_time])

# 运行 Agent
result = agent.run("What's the weather today and what time is it?")
print(result)

class MultiAgentSystem:
    def __init__(self):
        self.agents = []
        self.client = openai.OpenAI()
    
    def add_agent(self, agent: Agent):
        self.agents.append(agent)
    
    def coordinate(self, task: str) -> str:
        # 分析任务并分配给合适的 Agent
        task_analysis = self._analyze_task(task)
        
        # 找到合适的 Agent
        suitable_agents = self._find_suitable_agents(task_analysis)
        
        # 让 Agent 协作完成任务
        results = []
        for agent in suitable_agents:
            result = agent.run(task)
            results.append({
                "agent": agent.name,
                "result": result
            })
        
        # 整合结果
        return self._integrate_results(task, results)
    
    def _analyze_task(self, task: str) -> Dict:
        # 分析任务类型和需求
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {
                    "role": "system",
                    "content": """Analyze the task and return a JSON with the following fields:
{
  "type": "task type",
  "requirements": ["required capabilities"],
  "complexity": "low/medium/high"
}"""
                },
                {"role": "user", "content": f"Task: {task}"}
            ]
        )
        
        return json.loads(response.choices[0].message.content)
    
    def _find_suitable_agents(self, task_analysis: Dict) -> List[Agent]:
        # 根据任务需求找到合适的 Agent
        suitable_agents = []
        
        for agent in self.agents:
            # 检查 Agent 是否有任务所需的工具
            agent_capabilities = [tool.__name__ for tool in agent.tools]
            required_capabilities = task_analysis["requirements"]
            
            if any(cap in agent_capabilities for cap in required_capabilities):
                suitable_agents.append(agent)
        
        return suitable_agents
    
    def _integrate_results(self, task: str, results: List[Dict]) -> str:
        # 整合多个 Agent 的结果
        results_summary = "\n".join([
            f"{result['agent']}: {result['result']}"
            for result in results
        ])
        
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {
                    "role": "system",
                    "content": "Integrate the results from multiple agents into a coherent response."
                },
                {
                    "role": "user",
                    "content": f"""Task: {task}

Results:
{results_summary}

Integrated response:"""
                }
            ]
        )
        
        return response.choices[0].message.content

# 使用
system = MultiAgentSystem()

# 添加不同能力的 Agent
research_agent = Agent("Researcher", [search_web])
calculation_agent = Agent("Calculator", [calculate])
time_agent = Agent("Timekeeper", [get_current_time])

system.add_agent(research_agent)
system.add_agent(calculation_agent)
system.add_agent(time_agent)

# 协作完成任务
result = system.coordinate("Calculate 10 + 20, then search for information about the result, and tell me the current time.")
print(result)

def truncate_context(context, max_tokens=4000):
    # 简单截断
    if len(context.split()) <= max_tokens:
        return context
    
    # 更智能的方式：保留重要部分
    sentences = context.split('. ')
    truncated = '. '.join(sentences[:len(sentences) // 2])
    return truncated + '.'

from functools import lru_cache

class CachedLLMClient:
    def __init__(self):
        self.client = openai.OpenAI()
    
    @lru_cache(maxsize=1000)
    def completion(self, prompt: str) -> str:
        return self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        ).choices[0].message.content
    
    def cheap_completion(self, prompt: str) -> str:
        # 使用更便宜的模型
        return self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}]
        ).choices[0].message.content

def structured_output(prompt: str, schema: Dict) -> Dict:
    # 使用 Pydantic 或类似工具
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": f"Return the output in the following JSON format:\n{json.dumps(schema)}"
            },
            {"role": "user", "content": prompt}
        ]
    )
    
    try:
        return json.loads(response.choices[0].message.content)
    except json.JSONDecodeError:
        # 处理解析错误
        return {"error": "Failed to parse output"}