用户将传递一个简单的函数,并期望一个带有文档字符串、类型提示和函数解释作为输出的改进函数 我也想要 tool_schema 作为输出
所有这些输出都应该使用 langchain 的结构化_json 功能
我希望它成为 FastAPI 端点,并且 Gemini URL 应该硬编码在代码中
使用 Claude 生成人工智能
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langchain_google_genai import GoogleGenerativeAI
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from typing import Dict, List, Optional
app = FastAPI()
# Gemini configuration
GEMINI_URL = "your-gemini-url-here"
llm = GoogleGenerativeAI(model="gemini-pro", temperature=0)
# Pydantic models for structured output
class CodeAnalysis(BaseModel):
docstring: str
type_hints: Dict[str, str]
return_type: str
description: str
steps: List[str]
tool_schema: Dict[str, Any]
class FunctionRequest(BaseModel):
code: str
class FunctionResponse(BaseModel):
enhanced_code: str
analysis: CodeAnalysis
error: Optional[str] = None
# Create parser and prompt
parser = PydanticOutputParser(pydantic_object=CodeAnalysis)
prompt = PromptTemplate(
template="""Analyze this Python function and provide:
Function:
{code}
{format_instructions}
""",
input_variables=["code"],
partial_variables={"format_instructions": parser.get_format_instructions()}
)
@app.post("/analyze-function")
async def analyze_function(request: FunctionRequest) -> FunctionResponse:
try:
# Get LLM response
chain = prompt | llm | parser
analysis = chain.invoke({"code": request.code})
# Create enhanced code with docstring and type hints
enhanced_code = f"""def {request.code.split('def ')[1].split('(')[0]}({', '.join(f'{k}: {v}' for k, v in analysis.type_hints.items())}) -> {analysis.return_type}:
\"\"\"{analysis.docstring}\"\"\"
{request.code.split(':')[1]}
"""
return FunctionResponse(
enhanced_code=enhanced_code,
analysis=analysis,
error=None
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
使用 deepseek v3 生成的答案
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, validator
from typing import Dict, List, Optional
from langchain.chains import create_structured_output_chain
from langchain.prompts import ChatPromptTemplate
import inspect
app = FastAPI()
# Pydantic Models for Type Validation
class FunctionRequest(BaseModel):
code: str
@validator('code')
def validate_python_code(cls, code: str) -> str:
try:
compile(code, '<string>', 'exec')
return code
except SyntaxError:
raise ValueError("Invalid Python code")
class TypeHints(BaseModel):
parameters: Dict[str, str]
return_type: str = Field(default="Any")
class EnhancedFunctionOutput(BaseModel):
docstring: str
type_hints: TypeHints
description: str
steps: List[str]
name: str
class FunctionResponse(BaseModel):
enhanced_code: str
documentation: EnhancedFunctionOutput
error: Optional[str] = None
class FunctionAnalyzer:
def __init__(self, llm):
self.llm = llm
self.prompt = ChatPromptTemplate.from_template("""
Analyze this Python function and provide detailed documentation.
Function:
{code}
Provide:
1. Function name
2. Google-style docstring
3. Parameter types and return type
4. Brief description
5. Step-by-step breakdown of the function's operation
Be precise and technical.
""")
self.chain = create_structured_output_chain(
output_schema=EnhancedFunctionOutput,
llm=self.llm,
prompt=self.prompt,
)
def _enhance_function(self, code: str, docs: EnhancedFunctionOutput) -> str:
param_list = [
f"{name}: {type_}"
for name, type_ in docs.type_hints.parameters.items()
]
signature = f"def {docs.name}({', '.join(param_list)}) -> {docs.type_hints.return_type}"
body = inspect.getsource(compile(code, '<string>', 'exec')).split('\n', 1)[1]
return f"""{signature}:
\"\"\"{docs.docstring}\"\"\"
{body}"""
def analyze(self, code: str) -> FunctionResponse:
try:
# Get documentation from LLM
docs = self.chain.run(code=code)
# Create enhanced code
enhanced_code = self._enhance_function(code, docs)
return FunctionResponse(
enhanced_code=enhanced_code,
documentation=docs
)
except Exception as e:
return FunctionResponse(
enhanced_code=code,
documentation=None,
error=str(e)
)
# Hardcoded Gemini URL
GEMINI_URL = "https://api.gemini.com/v1" # Replace with actual URL
GEMINI_MODEL = None # Replace with actual Gemini model configuration
@app.post("/document-function")
async def document_function(request: FunctionRequest) -> FunctionResponse:
try:
analyzer = FunctionAnalyzer(GEMINI_MODEL) # Your configured Gemini model
return analyzer.analyze(request.code)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))