在实际技术项目中我们经常需要将AI能力集成到现有系统或构建新的智能应用。无论是使用大模型API、部署本地模型还是开发AI Agent一个清晰、可落地的工程路径都至关重要。本文旨在为开发者提供一个从零开始的AI应用开发实战指南我们将围绕一个具体的项目——构建一个具备基础交互能力的“AI小镇”模拟环境——来展开。这个过程会涉及环境搭建、依赖管理、核心代码编写、本地模型集成、基础Agent逻辑实现以及最终的效果验证和常见问题排查。无论你是希望了解如何将Spring AI等框架用起来还是想学习如何管理一个包含AI组件的完整项目都可以通过本文获得一个可复现的实践案例。1. 理解AI应用开发的核心组件与项目选型在开始编码之前我们需要明确构建一个AI应用通常涉及哪些部分以及如何为我们的“AI小镇”项目选择合适的技术栈。1.1 AI应用的核心分层一个典型的AI应用尤其是包含自主Agent或复杂交互的应用其架构可以粗略分为以下几层基础设施层提供计算资源。对于学习和小型项目个人电脑或一台云服务器足够对于需要运行大模型的场景则需要考虑GPU资源。本文项目对算力要求不高普通开发环境即可。模型服务层提供AI能力。可以是直接调用云端大模型API如OpenAI GPT、通义千问也可以在本地部署开源模型如Llama、ChatGLM。选择本地模型能更好地控制数据隐私和成本但需要处理模型部署和推理环境。应用框架层用于快速集成AI能力。例如Spring AI为Java开发者提供了统一的模型抽象和便捷的模板LangChainPython则擅长构建由LLM驱动的链式应用。考虑到项目“my_ai_town”可能是一个模拟游戏或社交实验我们选择灵活性较高的Python生态并会用到类似AI Agent的概念。业务逻辑层实现具体的应用功能。在“AI小镇”中这可能包括虚拟居民的日常行为决策、环境状态更新、事件触发与响应等。持久化与状态层存储Agent的记忆、环境状态、用户配置等。简单的项目可以使用文件如JSON或SQLite复杂的则需要数据库。1.2 项目“my_ai_town”的技术栈决策根据输入材料中提到的项目开源链接和关键词如ai agent,本地模型我们为这个实战项目确定以下技术选型编程语言Python。因其在AI和数据科学领域的丰富生态如langchain,transformers而成为首选。核心框架我们将使用LangChain的核心概念来构建Agent但会简化其实现以更直观地展示原理。同时会用到OpenAI风格的API与本地模型通信。本地模型服务为了模拟一个可离线运行、数据私有的“小镇”我们选择在本地部署一个轻量级开源大模型。这里选用ChatGLM3-6B的量化版本它相对较小对硬件要求较低且支持通过类似OpenAI的API接口调用。项目结构一个清晰的Python项目结构包含模型服务脚本、Agent核心逻辑、环境模拟器、主运行入口和配置文件。依赖管理使用pip和requirements.txt。这个选型平衡了学习成本、实践意义和资源消耗能够让我们在普通开发机上完成一个具备自主交互能力的AI小镇原型。2. 环境准备与依赖配置一个稳定的环境是项目成功的第一步。本节将详细说明如何准备Python环境、安装必要依赖并启动本地模型服务。2.1 Python环境与项目初始化首先确保你的开发机已安装Python建议版本3.8-3.11。然后创建项目目录并初始化虚拟环境这是管理项目依赖的最佳实践。# 创建项目目录 mkdir my_ai_town cd my_ai_town # 创建虚拟环境以venv为例 python -m venv venv # 激活虚拟环境 # 在Windows上 venv\Scripts\activate # 在Linux/macOS上 source venv/bin/activate # 激活后命令行提示符前通常会出现 (venv) 标识接下来创建项目的基础结构文件和依赖声明文件。# 创建基础目录和文件 mkdir -p agents environment utils touch main.py config.yaml utils/__init__.py agents/__init__.py environment/__init__.py # 创建依赖文件 requirements.txt touch requirements.txt2.2 安装核心Python依赖编辑requirements.txt文件加入以下内容。这些库涵盖了HTTP请求、配置管理、日期时间处理以及与大模型交互的核心组件。# requirements.txt fastapi0.104.0 uvicorn0.24.0 pydantic2.0.0 requests2.31.0 pyyaml6.0 python-dotenv1.0.0 openai1.0.0 # 注意这是OpenAI官方的新版Python SDK支持自定义base_url以连接本地模型 langchain0.0.340 # 我们将主要使用其提供的思路和部分工具而非复杂框架安装依赖pip install -r requirements.txt注意openai库版本1.0.0之后变化较大其ChatCompletion接口已被新的client.chat.completions.create方式取代。我们将使用新版SDK。2.3 部署本地大模型服务要让我们的AI小镇居民“有脑子”需要一个本地运行的大模型。我们使用ChatGLM3-6B并通过OpenAI兼容的API来调用它。步骤1下载模型可以从Hugging Face Model Hub下载THUDM/chatglm3-6b的4位量化版本chatglm3-6b-int4这能显著减少显存占用。你需要安装git-lfs来拉取大文件。步骤2使用开源API服务框架有许多项目可以将Hugging Face模型包装成OpenAI兼容的API服务。这里我们使用一个简单高效的方案fastchatvLLM或TGI(Text Generation Inference)。为了简化我们使用一个更轻量的示例脚本。在你的项目根目录创建一个local_model_server.py文件# local_model_server.py # 这是一个高度简化的示例实际生产应使用更稳定的框架如vLLM或TGI。 import uvicorn from fastapi import FastAPI from pydantic import BaseModel from transformers import AutoTokenizer, AutoModelForCausalLM import torch import warnings warnings.filterwarnings(ignore) app FastAPI(titleLocal ChatGLM3 API) # 加载模型和分词器 (请根据你的实际模型路径修改) MODEL_PATH ./models/chatglm3-6b-int4 # 假设模型已下载至此路径 tokenizer AutoTokenizer.from_pretrained(MODEL_PATH, trust_remote_codeTrue) model AutoModelForCausalLM.from_pretrained( MODEL_PATH, torch_dtypetorch.float16, # 半精度加载以节省显存 device_mapauto, # 自动分配设备CPU/GPU trust_remote_codeTrue ) model.eval() class ChatCompletionRequest(BaseModel): model: str chatglm3-6b messages: list max_tokens: int 512 temperature: float 0.7 app.post(/v1/chat/completions) async def create_chat_completion(request: ChatCompletionRequest): 模拟OpenAI的ChatCompletion接口。 # 将消息列表转换为ChatGLM3所需的prompt格式 prompt for msg in request.messages: role msg.get(role) content msg.get(content) if role system: prompt f[系统指令] {content}\n elif role user: prompt f[用户] {content}\n elif role assistant: prompt f[助手] {content}\n prompt [助手] # 生成回复 inputs tokenizer(prompt, return_tensorspt).to(model.device) with torch.no_grad(): outputs model.generate( **inputs, max_new_tokensrequest.max_tokens, temperaturerequest.temperature, do_sampleTrue if request.temperature 0 else False, pad_token_idtokenizer.eos_token_id ) response_text tokenizer.decode(outputs[0][inputs[input_ids].shape[1]:], skip_special_tokensTrue) # 构造OpenAI兼容的响应格式 return { id: chatcmpl-local, object: chat.completion, created: 1677652288, model: request.model, choices: [{ index: 0, message: { role: assistant, content: response_text.strip() }, finish_reason: length }], usage: { prompt_tokens: inputs[input_ids].shape[1], completion_tokens: len(outputs[0]) - inputs[input_ids].shape[1], total_tokens: len(outputs[0]) } } if __name__ __main__: # 启动服务默认监听本地8000端口 uvicorn.run(app, host0.0.0.0, port8000)重要说明此脚本仅为教学演示用于说明原理。在实际项目中强烈建议使用vLLM或TGI等经过优化的推理服务器它们能提供更高的吞吐量、更低的延迟以及完善的API兼容性。运行此脚本前你需要先使用pip install transformers torch安装相关库并确保有足够的GPU内存约6-8GB来加载INT4量化模型。步骤3启动模型服务# 在另一个终端窗口中激活虚拟环境后运行 python local_model_server.py如果一切正常你将看到服务启动日志并且可以通过http://localhost:8000/v1/chat/completions访问API。3. 构建AI小镇的核心模块我们的AI小镇将由几个核心模块构成配置管理、环境模拟、AI Agent定义以及主循环。我们将采用面向对象的设计让代码结构更清晰。3.1 项目配置与常量定义首先创建config.yaml文件用于集中管理配置项如模型API地址、Agent参数等。# config.yaml model: api_base: http://localhost:8000/v1 # 本地模型服务的地址 api_key: no-key-required # 本地服务通常不需要key但SDK要求有值 model_name: chatglm3-6b max_tokens: 256 temperature: 0.8 agents: default_memory_size: 5 # 每个Agent保留最近几条记忆 possible_actions: # Agent可以执行的基础动作 - 移动到[地点] - 与[人物]交谈 - 在[地点]工作 - 休息 - 思考[主题] environment: locations: [广场, 咖啡馆, 图书馆, 公园, 家] time_step_minutes: 30 # 每次世界推进的时间间隔分钟然后创建一个config.py模块来加载这些配置。# config.py import yaml import os from pydantic import BaseModel from typing import List class ModelConfig(BaseModel): api_base: str api_key: str model_name: str max_tokens: int temperature: float class AgentsConfig(BaseModel): default_memory_size: int possible_actions: List[str] class EnvironmentConfig(BaseModel): locations: List[str] time_step_minutes: int class Config(BaseModel): model: ModelConfig agents: AgentsConfig environment: EnvironmentConfig def load_config(config_path: str config.yaml) - Config: with open(config_path, r, encodingutf-8) as f: config_dict yaml.safe_load(f) return Config(**config_dict) # 全局配置对象 cfg load_config()3.2 定义AI Agent类Agent是小镇的居民它具有身份、记忆并能根据环境信息做出决策。在agents/目录下创建base_agent.py。# agents/base_agent.py import uuid from typing import List, Dict, Any from openai import OpenAI from config import cfg class BaseAgent: AI小镇的基础居民类。 每个Agent拥有独立的ID、名称、记忆并能通过LLM决定下一步行动。 def __init__(self, name: str, initial_location: str, traits: str): self.id str(uuid.uuid4())[:8] self.name name self.location initial_location self.traits traits # 性格特质描述用于塑造对话和行为风格 self.memory: List[str] [] # 记忆列表存储最近的事件 self.client OpenAI( base_urlcfg.model.api_base, api_keycfg.model.api_key ) def _update_memory(self, event: str): 更新记忆保留最近N条。 self.memory.append(event) if len(self.memory) cfg.agents.default_memory_size: self.memory.pop(0) def perceive(self, world_state: Dict[str, Any]) - str: 感知世界状态。这是一个简单的信息收集方法。 在实际复杂环境中这里可以过滤与Agent相关的信息。 # 当前只感知时间和地点 perceived f现在是{world_state[current_time]}我在{self.location}。 if self.memory: perceived f 我记得{, .join(self.memory[-2:])} # 加入最近两条记忆 return perceived def think_and_act(self, world_state: Dict[str, Any]) - str: 核心方法基于感知和记忆通过LLM思考并决定一个行动。 返回一个行动描述字符串。 perception self.perceive(world_state) prompt f你是一个名为{self.name}的虚拟小镇居民你的性格特点是{self.traits}。 当前情况 {perception} 你可以从以下行动中选择一个来执行{, .join(cfg.agents.possible_actions)}。 请只输出你选择的行动描述不要有任何额外的解释。例如“移动到图书馆”或“与朋友交谈”。 你的行动是 try: response self.client.chat.completions.create( modelcfg.model.model_name, messages[ {role: system, content: 你是一个虚拟小镇的居民根据当前情况和性格做出合理的行为选择。}, {role: user, content: prompt} ], max_tokenscfg.model.max_tokens, temperaturecfg.model.temperature ) action response.choices[0].message.content.strip() # 简单的行动后处理记录到记忆 self._update_memory(f在{world_state[current_time]}我{action}。) return action except Exception as e: # 如果模型调用失败返回一个默认行动 print(fAgent {self.name} 调用模型失败: {e}) return 休息 def __repr__(self): return fAgent({self.name}, 位置:{self.location})3.3 模拟小镇环境环境类负责维护世界状态如时间、地点列表并驱动每个时间步的更新。在environment/目录下创建world.py。# environment/world.py from datetime import datetime, timedelta from typing import List from agents.base_agent import BaseAgent from config import cfg class World: 小镇世界模拟器。 管理时间、地点和所有Agent并推进模拟。 def __init__(self): self.locations cfg.environment.locations self.current_time datetime.now().replace(hour8, minute0, second0) # 从早上8点开始 self.agents: List[BaseAgent] [] self.time_step timedelta(minutescfg.environment.time_step_minutes) self.log: List[str] [] def add_agent(self, agent: BaseAgent): 向世界添加一个Agent。 if agent.location not in self.locations: raise ValueError(f地点 {agent.location} 不在有效地点列表中。) self.agents.append(agent) self._log_event(f居民 {agent.name} 加入了小镇初始位置在 {agent.location}。) def get_state(self) - dict: 获取当前世界的摘要状态用于传递给Agent感知。 return { current_time: self.current_time.strftime(%H:%M), locations: self.locations, agent_count: len(self.agents) } def step(self): 推进一个时间步。 1. 每个Agent基于当前世界状态思考并行动。 2. 更新世界时间和日志。 world_state self.get_state() self._log_event(f--- 时间 {self.current_time.strftime(%H:%M)} ---) for agent in self.agents: action agent.think_and_act(world_state) # 这里可以解析action并实际更新agent的location等状态本例中仅记录 self._log_event(f{agent.name}: {action}) # 时间流逝 self.current_time self.time_step def _log_event(self, event: str): 记录事件到世界日志。 formatted_event f[{self.current_time.strftime(%H:%M)}] {event} print(formatted_event) self.log.append(formatted_event) def run_simulation(self, steps: int 10): 运行指定步数的模拟。 print(f开始模拟AI小镇共{steps}个时间步。) for i in range(steps): self.step() print(模拟结束。)4. 整合与运行让小镇活起来现在我们将所有模块整合起来创建一个主程序来初始化世界、添加居民并启动模拟。4.1 编写主程序入口在项目根目录创建main.py。# main.py from environment.world import World from agents.base_agent import BaseAgent def main(): # 1. 创建世界 world World() # 2. 创建一些具有不同性格的居民 alice BaseAgent(name爱丽丝, initial_location咖啡馆, traits热情开朗喜欢社交和咖啡。) bob BaseAgent(name鲍勃, initial_location图书馆, traits安静内向热爱读书和思考。) charlie BaseAgent(name查理, initial_location广场, traits精力充沛喜欢运动和到处走动。) # 3. 将居民添加到世界 for agent in [alice, bob, charlie]: world.add_agent(agent) # 4. 运行模拟例如模拟半天从8点到20点每半小时一步 world.run_simulation(steps24) # 5. 可选将日志保存到文件 with open(simulation_log.txt, w, encodingutf-8) as f: f.write(\n.join(world.log)) print(模拟日志已保存至 simulation_log.txt) if __name__ __main__: main()4.2 启动模拟与结果验证在启动前请确保你的本地模型服务local_model_server.py正在运行并且监听在http://localhost:8000。启动模型服务在终端Apython local_model_server.py等待看到Uvicorn running on http://0.0.0.0:8000的提示。运行AI小镇模拟在终端B确保已激活虚拟环境并在项目根目录python main.py如果一切顺利你将在终端B看到类似以下的输出这表示你的AI小镇正在运行居民们正在根据你的本地大模型“思考”并做出行动决策开始模拟AI小镇共24个时间步。 [08:00] 居民 爱丽丝 加入了小镇初始位置在 咖啡馆。 [08:00] 居民 鲍勃 加入了小镇初始位置在 图书馆。 [08:00] 居民 查理 加入了小镇初始位置在 广场。 --- 时间 08:00 --- 爱丽丝: 与鲍勃交谈 鲍勃: 在图书馆工作 查理: 移动到公园 --- 时间 08:30 --- 爱丽丝: 移动到图书馆 鲍勃: 思考哲学问题 查理: 在公园休息 ... 模拟结束。 模拟日志已保存至 simulation_log.txt这个输出表明集成成功你的Python程序成功通过HTTP请求调用了本地运行的ChatGLM3模型。Agent在运作每个Agent都基于其性格traits和当前感知到的世界状态生成了不同的行动。模拟在推进世界时间在按步前进事件被记录。5. 常见问题排查与优化在搭建和运行此类AI应用时你可能会遇到一些典型问题。下面是一个排查清单。5.1 模型服务连接失败问题现象可能原因检查方式处理建议openai.APIConnectionError或requests.exceptions.ConnectionError1. 模型服务未启动。2.config.yaml中的api_base地址或端口错误。3. 防火墙或网络策略阻止连接。1. 检查运行local_model_server.py的终端是否有错误。2. 在浏览器或使用curl http://localhost:8000/v1/models测试API端点。3. 检查config.yaml文件内容。1. 确保模型服务脚本正常运行且无报错。2. 将api_base改为正确的URL如http://127.0.0.1:8000/v1。3. 如果是云服务器检查安全组是否开放了对应端口。openai.AuthenticationError本地模型服务通常不需要API Key但新版OpenAI SDK要求该字段非空。检查config.yaml中api_key是否配置了一个非空字符串如no-key-required。确保config.yaml的model.api_key字段有值。5.2 模型响应异常或速度慢问题现象可能原因检查方式处理建议响应内容为乱码、无关或报错。1. 模型未正确加载或量化版本有问题。2. Prompt格式不符合模型要求。3. 显存不足导致推理出错。1. 查看模型服务终端的日志是否有加载错误或推理错误。2. 直接用简单Prompt如“你好”测试API。3. 使用nvidia-smi(GPU)或任务管理器查看内存占用。1. 确认模型文件完整尝试使用官方提供的示例代码加载测试。2. 调整local_model_server.py中的prompt转换逻辑使其符合ChatGLM3的对话格式。3. 尝试更小的量化版本如int8或使用CPU模式速度会慢很多。每个Agent行动决策耗时很长10秒。1. 本地模型推理本身较慢。2. 网络延迟虽然本地但仍有开销。3. 提示词过长导致生成缓慢。1. 观察模型服务终端的响应时间。2. 检查是否在循环中串行调用导致等待累积。1. 这是本地小模型的固有局限可考虑降低max_tokens或temperature。2. 如果未来Agent增多需要引入异步调用或队列。3. 精简Agent的记忆和prompt内容。5.3 项目运行与依赖问题问题现象可能原因检查方式处理建议ModuleNotFoundError1. 虚拟环境未激活。2.requirements.txt中的包未安装或版本冲突。1. 命令行前是否有(venv)标识。2. 运行pip list检查关键包是否存在。1. 激活正确的虚拟环境。2. 重新运行pip install -r requirements.txt注意查看警告信息。YAML解析错误或pydantic验证错误。config.yaml格式错误如缩进不对、键名错误。使用在线的YAML校验器检查config.yaml文件。确保YAML文件使用空格缩进通常是2个或4个空格键名与config.py中的Config类定义匹配。5.4 功能扩展与优化建议当前实现是一个极简的原型你可以从以下方向进行扩展使其更接近一个真正的“AI小镇”增强环境交互目前Agent的行动仅停留在文本描述。可以扩展World类使其能解析行动如“移动到[图书馆]”并实际更新Agent.location属性甚至定义地点之间的连接关系图结构。引入记忆与状态管理当前的记忆只是简单的字符串列表。可以设计更结构化的记忆体包含事件类型、时间、关联人物、情感权重等并实现基于向量数据库的长期记忆检索。实现Agent间通信当两个Agent处于同一地点时可以触发对话。这需要扩展think_and_act方法让Agent能接收其他Agent的发言作为输入并生成回复。对话历史也可以成为记忆的一部分。使用更稳定的模型服务将演示用的local_model_server.py替换为vLLM或TGI。它们提供生产级的API服务、动态批处理、流式输出等特性。加入前端界面使用Gradio、Streamlit或Web框架如FastAPI本身构建一个简单的可视化界面实时展示小镇地图、居民位置和对话气泡。配置外部模型如果想使用云端大模型如GPT-4只需修改config.yaml中的api_base为https://api.openai.com/v1并设置正确的api_key和model_name如gpt-4。代码无需改动这体现了通过配置和抽象接口带来的灵活性。通过这个项目你不仅实践了如何将本地大模型集成到Python应用中还初步构建了一个多Agent模拟系统的骨架。理解了这个基础框架后你可以继续探索更复杂的Agent架构、规划算法以及如何评估和优化这些智能体的行为这些都是当前AI Agent开发领域的前沿课题。 从零构建AI小镇：基于本地大模型与Agent的智能应用开发实战
在实际技术项目中我们经常需要将AI能力集成到现有系统或构建新的智能应用。无论是使用大模型API、部署本地模型还是开发AI Agent一个清晰、可落地的工程路径都至关重要。本文旨在为开发者提供一个从零开始的AI应用开发实战指南我们将围绕一个具体的项目——构建一个具备基础交互能力的“AI小镇”模拟环境——来展开。这个过程会涉及环境搭建、依赖管理、核心代码编写、本地模型集成、基础Agent逻辑实现以及最终的效果验证和常见问题排查。无论你是希望了解如何将Spring AI等框架用起来还是想学习如何管理一个包含AI组件的完整项目都可以通过本文获得一个可复现的实践案例。1. 理解AI应用开发的核心组件与项目选型在开始编码之前我们需要明确构建一个AI应用通常涉及哪些部分以及如何为我们的“AI小镇”项目选择合适的技术栈。1.1 AI应用的核心分层一个典型的AI应用尤其是包含自主Agent或复杂交互的应用其架构可以粗略分为以下几层基础设施层提供计算资源。对于学习和小型项目个人电脑或一台云服务器足够对于需要运行大模型的场景则需要考虑GPU资源。本文项目对算力要求不高普通开发环境即可。模型服务层提供AI能力。可以是直接调用云端大模型API如OpenAI GPT、通义千问也可以在本地部署开源模型如Llama、ChatGLM。选择本地模型能更好地控制数据隐私和成本但需要处理模型部署和推理环境。应用框架层用于快速集成AI能力。例如Spring AI为Java开发者提供了统一的模型抽象和便捷的模板LangChainPython则擅长构建由LLM驱动的链式应用。考虑到项目“my_ai_town”可能是一个模拟游戏或社交实验我们选择灵活性较高的Python生态并会用到类似AI Agent的概念。业务逻辑层实现具体的应用功能。在“AI小镇”中这可能包括虚拟居民的日常行为决策、环境状态更新、事件触发与响应等。持久化与状态层存储Agent的记忆、环境状态、用户配置等。简单的项目可以使用文件如JSON或SQLite复杂的则需要数据库。1.2 项目“my_ai_town”的技术栈决策根据输入材料中提到的项目开源链接和关键词如ai agent,本地模型我们为这个实战项目确定以下技术选型编程语言Python。因其在AI和数据科学领域的丰富生态如langchain,transformers而成为首选。核心框架我们将使用LangChain的核心概念来构建Agent但会简化其实现以更直观地展示原理。同时会用到OpenAI风格的API与本地模型通信。本地模型服务为了模拟一个可离线运行、数据私有的“小镇”我们选择在本地部署一个轻量级开源大模型。这里选用ChatGLM3-6B的量化版本它相对较小对硬件要求较低且支持通过类似OpenAI的API接口调用。项目结构一个清晰的Python项目结构包含模型服务脚本、Agent核心逻辑、环境模拟器、主运行入口和配置文件。依赖管理使用pip和requirements.txt。这个选型平衡了学习成本、实践意义和资源消耗能够让我们在普通开发机上完成一个具备自主交互能力的AI小镇原型。2. 环境准备与依赖配置一个稳定的环境是项目成功的第一步。本节将详细说明如何准备Python环境、安装必要依赖并启动本地模型服务。2.1 Python环境与项目初始化首先确保你的开发机已安装Python建议版本3.8-3.11。然后创建项目目录并初始化虚拟环境这是管理项目依赖的最佳实践。# 创建项目目录 mkdir my_ai_town cd my_ai_town # 创建虚拟环境以venv为例 python -m venv venv # 激活虚拟环境 # 在Windows上 venv\Scripts\activate # 在Linux/macOS上 source venv/bin/activate # 激活后命令行提示符前通常会出现 (venv) 标识接下来创建项目的基础结构文件和依赖声明文件。# 创建基础目录和文件 mkdir -p agents environment utils touch main.py config.yaml utils/__init__.py agents/__init__.py environment/__init__.py # 创建依赖文件 requirements.txt touch requirements.txt2.2 安装核心Python依赖编辑requirements.txt文件加入以下内容。这些库涵盖了HTTP请求、配置管理、日期时间处理以及与大模型交互的核心组件。# requirements.txt fastapi0.104.0 uvicorn0.24.0 pydantic2.0.0 requests2.31.0 pyyaml6.0 python-dotenv1.0.0 openai1.0.0 # 注意这是OpenAI官方的新版Python SDK支持自定义base_url以连接本地模型 langchain0.0.340 # 我们将主要使用其提供的思路和部分工具而非复杂框架安装依赖pip install -r requirements.txt注意openai库版本1.0.0之后变化较大其ChatCompletion接口已被新的client.chat.completions.create方式取代。我们将使用新版SDK。2.3 部署本地大模型服务要让我们的AI小镇居民“有脑子”需要一个本地运行的大模型。我们使用ChatGLM3-6B并通过OpenAI兼容的API来调用它。步骤1下载模型可以从Hugging Face Model Hub下载THUDM/chatglm3-6b的4位量化版本chatglm3-6b-int4这能显著减少显存占用。你需要安装git-lfs来拉取大文件。步骤2使用开源API服务框架有许多项目可以将Hugging Face模型包装成OpenAI兼容的API服务。这里我们使用一个简单高效的方案fastchatvLLM或TGI(Text Generation Inference)。为了简化我们使用一个更轻量的示例脚本。在你的项目根目录创建一个local_model_server.py文件# local_model_server.py # 这是一个高度简化的示例实际生产应使用更稳定的框架如vLLM或TGI。 import uvicorn from fastapi import FastAPI from pydantic import BaseModel from transformers import AutoTokenizer, AutoModelForCausalLM import torch import warnings warnings.filterwarnings(ignore) app FastAPI(titleLocal ChatGLM3 API) # 加载模型和分词器 (请根据你的实际模型路径修改) MODEL_PATH ./models/chatglm3-6b-int4 # 假设模型已下载至此路径 tokenizer AutoTokenizer.from_pretrained(MODEL_PATH, trust_remote_codeTrue) model AutoModelForCausalLM.from_pretrained( MODEL_PATH, torch_dtypetorch.float16, # 半精度加载以节省显存 device_mapauto, # 自动分配设备CPU/GPU trust_remote_codeTrue ) model.eval() class ChatCompletionRequest(BaseModel): model: str chatglm3-6b messages: list max_tokens: int 512 temperature: float 0.7 app.post(/v1/chat/completions) async def create_chat_completion(request: ChatCompletionRequest): 模拟OpenAI的ChatCompletion接口。 # 将消息列表转换为ChatGLM3所需的prompt格式 prompt for msg in request.messages: role msg.get(role) content msg.get(content) if role system: prompt f[系统指令] {content}\n elif role user: prompt f[用户] {content}\n elif role assistant: prompt f[助手] {content}\n prompt [助手] # 生成回复 inputs tokenizer(prompt, return_tensorspt).to(model.device) with torch.no_grad(): outputs model.generate( **inputs, max_new_tokensrequest.max_tokens, temperaturerequest.temperature, do_sampleTrue if request.temperature 0 else False, pad_token_idtokenizer.eos_token_id ) response_text tokenizer.decode(outputs[0][inputs[input_ids].shape[1]:], skip_special_tokensTrue) # 构造OpenAI兼容的响应格式 return { id: chatcmpl-local, object: chat.completion, created: 1677652288, model: request.model, choices: [{ index: 0, message: { role: assistant, content: response_text.strip() }, finish_reason: length }], usage: { prompt_tokens: inputs[input_ids].shape[1], completion_tokens: len(outputs[0]) - inputs[input_ids].shape[1], total_tokens: len(outputs[0]) } } if __name__ __main__: # 启动服务默认监听本地8000端口 uvicorn.run(app, host0.0.0.0, port8000)重要说明此脚本仅为教学演示用于说明原理。在实际项目中强烈建议使用vLLM或TGI等经过优化的推理服务器它们能提供更高的吞吐量、更低的延迟以及完善的API兼容性。运行此脚本前你需要先使用pip install transformers torch安装相关库并确保有足够的GPU内存约6-8GB来加载INT4量化模型。步骤3启动模型服务# 在另一个终端窗口中激活虚拟环境后运行 python local_model_server.py如果一切正常你将看到服务启动日志并且可以通过http://localhost:8000/v1/chat/completions访问API。3. 构建AI小镇的核心模块我们的AI小镇将由几个核心模块构成配置管理、环境模拟、AI Agent定义以及主循环。我们将采用面向对象的设计让代码结构更清晰。3.1 项目配置与常量定义首先创建config.yaml文件用于集中管理配置项如模型API地址、Agent参数等。# config.yaml model: api_base: http://localhost:8000/v1 # 本地模型服务的地址 api_key: no-key-required # 本地服务通常不需要key但SDK要求有值 model_name: chatglm3-6b max_tokens: 256 temperature: 0.8 agents: default_memory_size: 5 # 每个Agent保留最近几条记忆 possible_actions: # Agent可以执行的基础动作 - 移动到[地点] - 与[人物]交谈 - 在[地点]工作 - 休息 - 思考[主题] environment: locations: [广场, 咖啡馆, 图书馆, 公园, 家] time_step_minutes: 30 # 每次世界推进的时间间隔分钟然后创建一个config.py模块来加载这些配置。# config.py import yaml import os from pydantic import BaseModel from typing import List class ModelConfig(BaseModel): api_base: str api_key: str model_name: str max_tokens: int temperature: float class AgentsConfig(BaseModel): default_memory_size: int possible_actions: List[str] class EnvironmentConfig(BaseModel): locations: List[str] time_step_minutes: int class Config(BaseModel): model: ModelConfig agents: AgentsConfig environment: EnvironmentConfig def load_config(config_path: str config.yaml) - Config: with open(config_path, r, encodingutf-8) as f: config_dict yaml.safe_load(f) return Config(**config_dict) # 全局配置对象 cfg load_config()3.2 定义AI Agent类Agent是小镇的居民它具有身份、记忆并能根据环境信息做出决策。在agents/目录下创建base_agent.py。# agents/base_agent.py import uuid from typing import List, Dict, Any from openai import OpenAI from config import cfg class BaseAgent: AI小镇的基础居民类。 每个Agent拥有独立的ID、名称、记忆并能通过LLM决定下一步行动。 def __init__(self, name: str, initial_location: str, traits: str): self.id str(uuid.uuid4())[:8] self.name name self.location initial_location self.traits traits # 性格特质描述用于塑造对话和行为风格 self.memory: List[str] [] # 记忆列表存储最近的事件 self.client OpenAI( base_urlcfg.model.api_base, api_keycfg.model.api_key ) def _update_memory(self, event: str): 更新记忆保留最近N条。 self.memory.append(event) if len(self.memory) cfg.agents.default_memory_size: self.memory.pop(0) def perceive(self, world_state: Dict[str, Any]) - str: 感知世界状态。这是一个简单的信息收集方法。 在实际复杂环境中这里可以过滤与Agent相关的信息。 # 当前只感知时间和地点 perceived f现在是{world_state[current_time]}我在{self.location}。 if self.memory: perceived f 我记得{, .join(self.memory[-2:])} # 加入最近两条记忆 return perceived def think_and_act(self, world_state: Dict[str, Any]) - str: 核心方法基于感知和记忆通过LLM思考并决定一个行动。 返回一个行动描述字符串。 perception self.perceive(world_state) prompt f你是一个名为{self.name}的虚拟小镇居民你的性格特点是{self.traits}。 当前情况 {perception} 你可以从以下行动中选择一个来执行{, .join(cfg.agents.possible_actions)}。 请只输出你选择的行动描述不要有任何额外的解释。例如“移动到图书馆”或“与朋友交谈”。 你的行动是 try: response self.client.chat.completions.create( modelcfg.model.model_name, messages[ {role: system, content: 你是一个虚拟小镇的居民根据当前情况和性格做出合理的行为选择。}, {role: user, content: prompt} ], max_tokenscfg.model.max_tokens, temperaturecfg.model.temperature ) action response.choices[0].message.content.strip() # 简单的行动后处理记录到记忆 self._update_memory(f在{world_state[current_time]}我{action}。) return action except Exception as e: # 如果模型调用失败返回一个默认行动 print(fAgent {self.name} 调用模型失败: {e}) return 休息 def __repr__(self): return fAgent({self.name}, 位置:{self.location})3.3 模拟小镇环境环境类负责维护世界状态如时间、地点列表并驱动每个时间步的更新。在environment/目录下创建world.py。# environment/world.py from datetime import datetime, timedelta from typing import List from agents.base_agent import BaseAgent from config import cfg class World: 小镇世界模拟器。 管理时间、地点和所有Agent并推进模拟。 def __init__(self): self.locations cfg.environment.locations self.current_time datetime.now().replace(hour8, minute0, second0) # 从早上8点开始 self.agents: List[BaseAgent] [] self.time_step timedelta(minutescfg.environment.time_step_minutes) self.log: List[str] [] def add_agent(self, agent: BaseAgent): 向世界添加一个Agent。 if agent.location not in self.locations: raise ValueError(f地点 {agent.location} 不在有效地点列表中。) self.agents.append(agent) self._log_event(f居民 {agent.name} 加入了小镇初始位置在 {agent.location}。) def get_state(self) - dict: 获取当前世界的摘要状态用于传递给Agent感知。 return { current_time: self.current_time.strftime(%H:%M), locations: self.locations, agent_count: len(self.agents) } def step(self): 推进一个时间步。 1. 每个Agent基于当前世界状态思考并行动。 2. 更新世界时间和日志。 world_state self.get_state() self._log_event(f--- 时间 {self.current_time.strftime(%H:%M)} ---) for agent in self.agents: action agent.think_and_act(world_state) # 这里可以解析action并实际更新agent的location等状态本例中仅记录 self._log_event(f{agent.name}: {action}) # 时间流逝 self.current_time self.time_step def _log_event(self, event: str): 记录事件到世界日志。 formatted_event f[{self.current_time.strftime(%H:%M)}] {event} print(formatted_event) self.log.append(formatted_event) def run_simulation(self, steps: int 10): 运行指定步数的模拟。 print(f开始模拟AI小镇共{steps}个时间步。) for i in range(steps): self.step() print(模拟结束。)4. 整合与运行让小镇活起来现在我们将所有模块整合起来创建一个主程序来初始化世界、添加居民并启动模拟。4.1 编写主程序入口在项目根目录创建main.py。# main.py from environment.world import World from agents.base_agent import BaseAgent def main(): # 1. 创建世界 world World() # 2. 创建一些具有不同性格的居民 alice BaseAgent(name爱丽丝, initial_location咖啡馆, traits热情开朗喜欢社交和咖啡。) bob BaseAgent(name鲍勃, initial_location图书馆, traits安静内向热爱读书和思考。) charlie BaseAgent(name查理, initial_location广场, traits精力充沛喜欢运动和到处走动。) # 3. 将居民添加到世界 for agent in [alice, bob, charlie]: world.add_agent(agent) # 4. 运行模拟例如模拟半天从8点到20点每半小时一步 world.run_simulation(steps24) # 5. 可选将日志保存到文件 with open(simulation_log.txt, w, encodingutf-8) as f: f.write(\n.join(world.log)) print(模拟日志已保存至 simulation_log.txt) if __name__ __main__: main()4.2 启动模拟与结果验证在启动前请确保你的本地模型服务local_model_server.py正在运行并且监听在http://localhost:8000。启动模型服务在终端Apython local_model_server.py等待看到Uvicorn running on http://0.0.0.0:8000的提示。运行AI小镇模拟在终端B确保已激活虚拟环境并在项目根目录python main.py如果一切顺利你将在终端B看到类似以下的输出这表示你的AI小镇正在运行居民们正在根据你的本地大模型“思考”并做出行动决策开始模拟AI小镇共24个时间步。 [08:00] 居民 爱丽丝 加入了小镇初始位置在 咖啡馆。 [08:00] 居民 鲍勃 加入了小镇初始位置在 图书馆。 [08:00] 居民 查理 加入了小镇初始位置在 广场。 --- 时间 08:00 --- 爱丽丝: 与鲍勃交谈 鲍勃: 在图书馆工作 查理: 移动到公园 --- 时间 08:30 --- 爱丽丝: 移动到图书馆 鲍勃: 思考哲学问题 查理: 在公园休息 ... 模拟结束。 模拟日志已保存至 simulation_log.txt这个输出表明集成成功你的Python程序成功通过HTTP请求调用了本地运行的ChatGLM3模型。Agent在运作每个Agent都基于其性格traits和当前感知到的世界状态生成了不同的行动。模拟在推进世界时间在按步前进事件被记录。5. 常见问题排查与优化在搭建和运行此类AI应用时你可能会遇到一些典型问题。下面是一个排查清单。5.1 模型服务连接失败问题现象可能原因检查方式处理建议openai.APIConnectionError或requests.exceptions.ConnectionError1. 模型服务未启动。2.config.yaml中的api_base地址或端口错误。3. 防火墙或网络策略阻止连接。1. 检查运行local_model_server.py的终端是否有错误。2. 在浏览器或使用curl http://localhost:8000/v1/models测试API端点。3. 检查config.yaml文件内容。1. 确保模型服务脚本正常运行且无报错。2. 将api_base改为正确的URL如http://127.0.0.1:8000/v1。3. 如果是云服务器检查安全组是否开放了对应端口。openai.AuthenticationError本地模型服务通常不需要API Key但新版OpenAI SDK要求该字段非空。检查config.yaml中api_key是否配置了一个非空字符串如no-key-required。确保config.yaml的model.api_key字段有值。5.2 模型响应异常或速度慢问题现象可能原因检查方式处理建议响应内容为乱码、无关或报错。1. 模型未正确加载或量化版本有问题。2. Prompt格式不符合模型要求。3. 显存不足导致推理出错。1. 查看模型服务终端的日志是否有加载错误或推理错误。2. 直接用简单Prompt如“你好”测试API。3. 使用nvidia-smi(GPU)或任务管理器查看内存占用。1. 确认模型文件完整尝试使用官方提供的示例代码加载测试。2. 调整local_model_server.py中的prompt转换逻辑使其符合ChatGLM3的对话格式。3. 尝试更小的量化版本如int8或使用CPU模式速度会慢很多。每个Agent行动决策耗时很长10秒。1. 本地模型推理本身较慢。2. 网络延迟虽然本地但仍有开销。3. 提示词过长导致生成缓慢。1. 观察模型服务终端的响应时间。2. 检查是否在循环中串行调用导致等待累积。1. 这是本地小模型的固有局限可考虑降低max_tokens或temperature。2. 如果未来Agent增多需要引入异步调用或队列。3. 精简Agent的记忆和prompt内容。5.3 项目运行与依赖问题问题现象可能原因检查方式处理建议ModuleNotFoundError1. 虚拟环境未激活。2.requirements.txt中的包未安装或版本冲突。1. 命令行前是否有(venv)标识。2. 运行pip list检查关键包是否存在。1. 激活正确的虚拟环境。2. 重新运行pip install -r requirements.txt注意查看警告信息。YAML解析错误或pydantic验证错误。config.yaml格式错误如缩进不对、键名错误。使用在线的YAML校验器检查config.yaml文件。确保YAML文件使用空格缩进通常是2个或4个空格键名与config.py中的Config类定义匹配。5.4 功能扩展与优化建议当前实现是一个极简的原型你可以从以下方向进行扩展使其更接近一个真正的“AI小镇”增强环境交互目前Agent的行动仅停留在文本描述。可以扩展World类使其能解析行动如“移动到[图书馆]”并实际更新Agent.location属性甚至定义地点之间的连接关系图结构。引入记忆与状态管理当前的记忆只是简单的字符串列表。可以设计更结构化的记忆体包含事件类型、时间、关联人物、情感权重等并实现基于向量数据库的长期记忆检索。实现Agent间通信当两个Agent处于同一地点时可以触发对话。这需要扩展think_and_act方法让Agent能接收其他Agent的发言作为输入并生成回复。对话历史也可以成为记忆的一部分。使用更稳定的模型服务将演示用的local_model_server.py替换为vLLM或TGI。它们提供生产级的API服务、动态批处理、流式输出等特性。加入前端界面使用Gradio、Streamlit或Web框架如FastAPI本身构建一个简单的可视化界面实时展示小镇地图、居民位置和对话气泡。配置外部模型如果想使用云端大模型如GPT-4只需修改config.yaml中的api_base为https://api.openai.com/v1并设置正确的api_key和model_name如gpt-4。代码无需改动这体现了通过配置和抽象接口带来的灵活性。通过这个项目你不仅实践了如何将本地大模型集成到Python应用中还初步构建了一个多Agent模拟系统的骨架。理解了这个基础框架后你可以继续探索更复杂的Agent架构、规划算法以及如何评估和优化这些智能体的行为这些都是当前AI Agent开发领域的前沿课题。