金融行业LLM应用落地：受限环境下的ReAct架构实践

1. 企业级LLM应用落地的核心挑战

在金融行业IT环境中部署大型语言模型（LLM）应用时，我们面临着独特的限制条件。以某商业银行的智能助手项目为例，技术团队只能使用内部提供的Copilot API，这个API虽然基于主流LLM架构，但进行了深度定制和功能裁剪。最显著的限制是：标准Function Calling接口（如OpenAI的bind_tools）被完全移除，这使得常规的Agent构建方法失效。

1.1 金融IT环境的特殊约束

银行系统的技术架构通常具有以下特征：

• 网络隔离：生产环境与互联网物理隔离，无法直接调用外部API
• 审计要求：所有AI生成内容需要完整留痕，包括中间思考过程
• 性能约束：响应延迟必须控制在3秒内完成复杂任务
• 协议限制：仅允许使用经过安全审查的特定通信协议

1.2 Copilot API的功能分析

我们拿到的内部Copilot API具有以下能力边界：

python复制class CopilotAPI:
    async def ainvoke(self, prompt: str) -> str:  # 基础文本生成
    async def astream(self, prompt: str) -> AsyncGenerator:  # 流式输出
    # 缺失的关键功能：
    # - bind_tools()
    # - function_calling()
    # - structured_output()

2. 架构设计与技术选型

2.1 整体架构方案

我们采用分层架构设计，各组件职责明确：

code复制┌───────────────────────┐
│      前端界面层        │
└──────────┬────────────┘
           │ HTTP/HTTPS
┌──────────▼────────────┐
│    API网关层           │
│  (请求路由/权限校验)    │
└──────────┬────────────┘
           │ gRPC
┌──────────▼────────────┐
│    Agent协调层         │
│  (MainAgent+GithubAgent)│
└──────────┬────────────┘
           │ Protobuf
┌──────────▼────────────┐
│   工具执行层           │
│  (MCP协议适配器)       │
└──────────┬────────────┘
           │ REST
┌──────────▼────────────┐
│   Copilot API服务      │
└───────────────────────┘

2.2 关键技术决策

2.2.1 选择ReAct模式的原因

在标准Function Calling不可用的情况下，ReAct（Reasoning+Acting）模式成为最优选择：

1. 兼容性强：仅需模型具备基础文本生成能力
2. 可解释性：保留完整的思考链（Chain-of-Thought）
3. 灵活性：可自定义交互协议

2.2.2 Model Context Protocol设计

我们设计了专用的MCP协议规范：

json复制{
  "version": "1.0",
  "tools": [
    {
      "name": "get_repo_list",
      "description": "获取用户GitHub仓库列表",
      "inputSchema": {
        "type": "object",
        "properties": {
          "username": {
            "type": "string",
            "description": "GitHub用户名"
          }
        }
      }
    }
  ]
}

3. 核心组件实现细节

3.1 McpToolConverter的实现

工具协议转换器的完整实现包含以下关键逻辑：

python复制from pydantic import create_model, Field
from langchain.tools import StructuredTool

class McpToolConverter:
    @staticmethod
    def convert(tool: dict) -> StructuredTool:
        # 动态生成参数模型
        fields = {}
        for param, schema in tool["inputSchema"]["properties"].items():
            field_type = {
                "string": str,
                "number": float,
                "integer": int,
                "boolean": bool
            }[schema["type"]]
            
            fields[param] = (
                field_type,
                Field(..., description=schema.get("description", ""))
            )
        
        args_model = create_model(
            f"{tool['name']}Model",
            **fields
        )
        
        # 实际工具函数
        async def tool_func(**kwargs):
            # 这里会调用实际的MCP服务
            return await call_mcp_service(tool["name"], kwargs)
            
        return StructuredTool.from_function(
            func=tool_func,
            name=tool["name"],
            description=tool["description"],
            args_schema=args_model
        )

3.2 ToolCallableAgent的Prompt工程

系统提示词的设计直接影响模型行为，我们的模板包含以下关键部分：

python复制def build_system_prompt(tools: List[dict]) -> str:
    tool_descriptions = []
    for tool in tools:
        args_desc = "\n".join(
            f"- {name}: {schema['type']} ({schema.get('description','')})"
            for name, schema in tool["inputSchema"]["properties"].items()
        )
        tool_descriptions.append(
            f"{tool['name']}:\n"
            f"  功能: {tool['description']}\n"
            f"  参数:\n{args_desc}"
        )
    
    return f"""你是一个专业助手，可以调用以下工具：
    
{'\n\n'.join(tool_descriptions)}

调用工具时，请严格使用以下JSON格式（包含markdown代码块）：
```json
{{
    "action": "工具名称",
    "action_input": {{"参数名": "参数值"}}
}}

重要规则：

1. 每次只能调用一个工具
2. 必须提供所有必需参数
3. 等待工具返回结果后再继续
   """

code复制
## 4. 关键问题与解决方案

### 4.1 流式输出的中断处理

在实现实时流式输出时，我们发现当模型开始输出JSON工具调用时，后续文本可能包含干扰内容。解决方案：

```python
class OutputInterceptor:
    def __init__(self):
        self.buffer = []
        self.in_json_block = False
    
    async def intercept(self, chunk: str) -> Optional[dict]:
        self.buffer.append(chunk)
        full_text = "".join(self.buffer)
        
        # 检测JSON块开始
        if "```json" in full_text and not self.in_json_block:
            self.in_json_block = True
            return None
        
        # 提取完整JSON
        if self.in_json_block and "```" in full_text:
            json_part = full_text.split("```json")[1].split("```")[0]
            try:
                return json.loads(json_part)
            except JSONDecodeError:
                pass
        
        return None

4.2 历史记录去重算法

为解决重复处理用户输入的问题，我们实现了以下去重逻辑：

python复制def deduplicate_history(history: List[dict], new_input: str) -> List[dict]:
    if not history:
        return []
    
    # 检查最后一条是否与当前输入相同
    last_msg = history[-1]
    if last_msg["role"] == "user" and last_msg["content"] == new_input:
        return history[:-1]
    
    # 检查连续重复问题
    if len(history) >= 2:
        prev_msg = history[-2]
        if (prev_msg["role"] == "assistant" and 
            "请提供" in prev_msg["content"] and 
            last_msg["role"] == "user"):
            return history[:-2]
    
    return history

5. 性能优化实践

5.1 工具调用并行化

虽然ReAct要求顺序执行，但我们可以在某些场景下优化：

python复制async def execute_tools(tool_calls: List[dict]) -> List[dict]:
    # 分类工具调用：独立工具可并行执行
    independent_tools = []
    dependent_tools = []
    
    for call in tool_calls:
        if is_independent_tool(call["action"]):
            independent_tools.append(call)
        else:
            dependent_tools.append(call)
    
    # 并行执行独立工具
    independent_results = await asyncio.gather(
        *[call_tool(t["action"], t["action_input"]) 
          for t in independent_tools]
    )
    
    # 顺序执行依赖工具
    dependent_results = []
    for tool in dependent_tools:
        res = await call_tool(tool["action"], tool["action_input"])
        dependent_results.append(res)
    
    return independent_results + dependent_results

5.2 缓存策略实现

针对高频工具调用，我们设计了双层缓存：

python复制class ToolCache:
    def __init__(self):
        self.memory_cache = {}
        self.redis_pool = redis.ConnectionPool()
    
    async def get(self, tool_name: str, params: dict) -> Any:
        cache_key = self._generate_key(tool_name, params)
        
        # 第一层：内存缓存
        if cache_key in self.memory_cache:
            return self.memory_cache[cache_key]
        
        # 第二层：Redis缓存
        redis_client = redis.Redis(connection_pool=self.redis_pool)
        cached = await redis_client.get(cache_key)
        if cached:
            result = json.loads(cached)
            self.memory_cache[cache_key] = result
            return result
        
        # 实际调用
        result = await call_tool(tool_name, params)
        
        # 缓存结果（根据工具特性设置不同TTL）
        ttl = self._get_ttl(tool_name)
        await redis_client.setex(cache_key, ttl, json.dumps(result))
        self.memory_cache[cache_key] = result
        
        return result

6. 安全合规措施

6.1 输出内容过滤

在金融场景下，所有输出必须经过安全过滤：

python复制class ContentFilter:
    def __init__(self):
        self.sensitive_keywords = load_keywords("sensitive_words.txt")
    
    async def filter(self, text: str) -> str:
        # 关键词过滤
        for word in self.sensitive_keywords:
            if word in text.lower():
                text = text.replace(word, "***")
        
        # PII数据脱敏
        text = self._mask_pii(text)
        
        # 代码注入防护
        text = self._sanitize_code(text)
        
        return text
    
    def _mask_pii(self, text: str) -> str:
        # 使用正则表达式匹配身份证号、银行卡号等
        patterns = {
            r"\b\d{17}[\dXx]\b": "ID_NUMBER",
            r"\b\d{16}\b": "CARD_NUMBER"
        }
        for pattern, replacement in patterns.items():
            text = re.sub(pattern, replacement, text)
        return text

6.2 审计日志实现

完整的审计日志包含以下信息：

python复制class AuditLogger:
    async def log_interaction(self, session_id: str, data: dict):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "session_id": session_id,
            "user_input": data["input"],
            "full_chain": data["chain"],
            "tool_calls": data.get("tools", []),
            "final_output": data["output"],
            "metadata": {
                "response_time": data["response_time"],
                "model_version": data["model_version"]
            }
        }
        
        # 写入Elasticsearch
        await self.es_client.index(
            index="llm_audit_logs",
            document=log_entry
        )
        
        # 同时写入区块链存证
        if is_sensitive_interaction(data):
            await self.blockchain_client.add_entry(
                collection="audit_trail",
                entry=log_entry
            )

7. 部署与监控方案

7.1 Kubernetes部署配置

我们的生产环境部署使用以下关键配置：

yaml复制# deployment.yaml关键部分
resources:
  limits:
    cpu: "2"
    memory: "4Gi"
  requests:
    cpu: "1"
    memory: "2Gi"

readinessProbe:
  httpGet:
    path: /health
    port: 8000
  initialDelaySeconds: 10
  periodSeconds: 5

livenessProbe:
  exec:
    command: ["python", "healthcheck.py"]
  initialDelaySeconds: 30
  periodSeconds: 10

7.2 监控指标设计

我们跟踪的核心指标包括：

8. 实际效果与业务价值

8.1 性能基准测试

经过优化后的系统表现：

8.2 典型业务场景

场景一：内部知识库查询

text复制用户：查找2023年外汇风险管理政策
Agent：
1. 思考：需要查询文档管理系统
2. 调用document_search工具
3. 分析返回的3个相关文档
4. 生成摘要回答

场景二：开发支持

text复制用户：我的PR为什么构建失败？
Agent：
1. 识别需要GitHub信息
2. 调用get_pr_details工具
3. 分析构建日志
4. 定位到单元测试失败
5. 建议修复方法

9. 经验总结与改进方向

在实际落地过程中，我们积累了以下关键经验：

1. 渐进式复杂度控制：先实现单工具调用，再扩展多工具协作
2. 强类型验证：对所有工具参数实施严格的Pydantic验证
3. 回退机制：当JSON解析失败时自动切换纯文本模式
4. 用户引导：当检测到可能工具调用时主动提示用户提供必要参数

未来改进方向：

• 引入更强大的意图识别模块
• 实现工具用法的自动学习机制
• 开发可视化调试工具
• 优化上下文窗口使用效率